datajoint-python/tests/test_alter.py at a30d41be1b4ea287dd20933546397b7cd8fbcd65 · d-v-b/datajoint-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import re

import pytest

import datajoint as dj

from . import schema as schema_any_module
from .schema_alter import LOCALS_ALTER, Experiment, Parent

COMBINED_CONTEXT = {
    **schema_any_module.LOCALS_ANY,
    **LOCALS_ALTER,
}


@pytest.fixture
def schema_alter(connection_test, schema_any_fresh):
    # Overwrite Experiment and Parent nodes using fresh schema
    schema_any_fresh(Experiment, context=LOCALS_ALTER)
    schema_any_fresh(Parent, context=LOCALS_ALTER)
    yield schema_any_fresh
    schema_any_fresh.drop()


class TestAlter:
    def verify_alter(self, schema_alter, table, attribute_sql):
        definition_original = schema_alter.connection.query(
            f"SHOW CREATE TABLE {table.full_table_name}"
        ).fetchone()[1]
        table.definition = table.definition_new
        table.alter(prompt=False)
        definition_new = schema_alter.connection.query(
            f"SHOW CREATE TABLE {table.full_table_name}"
        ).fetchone()[1]
        assert (
            re.sub(f"{attribute_sql},\n  ", "", definition_new) == definition_original
        )

    def test_alter(self, schema_alter):
        original = schema_alter.connection.query(
            "SHOW CREATE TABLE " + Experiment.full_table_name
        ).fetchone()[1]
        Experiment.definition = Experiment.definition1
        Experiment.alter(prompt=False, context=COMBINED_CONTEXT)
        altered = schema_alter.connection.query(
            "SHOW CREATE TABLE " + Experiment.full_table_name
        ).fetchone()[1]
        assert original != altered
        Experiment.definition = Experiment.original_definition
        Experiment().alter(prompt=False, context=COMBINED_CONTEXT)
        restored = schema_alter.connection.query(
            "SHOW CREATE TABLE " + Experiment.full_table_name
        ).fetchone()[1]
        assert altered != restored
        assert original == restored

    def test_alter_part(self, schema_alter):
        """
        https://github.com/datajoint/datajoint-python/issues/936
        """
        self.verify_alter(
            schema_alter, table=Parent.Child, attribute_sql="`child_id` .* DEFAULT NULL"
        )
        self.verify_alter(
            schema_alter,
            table=Parent.Grandchild,
            attribute_sql="`grandchild_id` .* DEFAULT NULL",
        )