|
| 1 | +import types |
| 2 | +import pytest |
| 3 | +import inspect |
| 4 | +import datajoint as dj |
| 5 | +from unittest.mock import patch |
| 6 | +from inspect import getmembers |
| 7 | +from . import schema |
| 8 | +from . import PREFIX |
| 9 | + |
| 10 | + |
| 11 | +class Ephys(dj.Imported): |
| 12 | + definition = """ # This is already declare in ./schema.py |
| 13 | + """ |
| 14 | + |
| 15 | + |
| 16 | +def relation_selector(attr): |
| 17 | + try: |
| 18 | + return issubclass(attr, dj.Table) |
| 19 | + except TypeError: |
| 20 | + return False |
| 21 | + |
| 22 | + |
| 23 | +def part_selector(attr): |
| 24 | + try: |
| 25 | + return issubclass(attr, dj.Part) |
| 26 | + except TypeError: |
| 27 | + return False |
| 28 | + |
| 29 | + |
| 30 | +@pytest.fixture |
| 31 | +def schema_empty_module(schema_any, schema_empty): |
| 32 | + """ |
| 33 | + Mock the module tests_old.schema_empty. |
| 34 | + The test `test_namespace_population` will check that the module contains all the |
| 35 | + classes in schema_any, after running `spawn_missing_classes`. |
| 36 | + """ |
| 37 | + namespace_dict = { |
| 38 | + "_": schema_any, |
| 39 | + "schema": schema_empty, |
| 40 | + "Ephys": Ephys, |
| 41 | + } |
| 42 | + module = types.ModuleType("schema_empty") |
| 43 | + |
| 44 | + # Add classes to the module's namespace |
| 45 | + for k, v in namespace_dict.items(): |
| 46 | + setattr(module, k, v) |
| 47 | + |
| 48 | + return module |
| 49 | + |
| 50 | + |
| 51 | +@pytest.fixture |
| 52 | +def schema_empty(connection_test, schema_any): |
| 53 | + context = {**schema.LOCALS_ANY, "Ephys": Ephys} |
| 54 | + schema_empty = dj.Schema( |
| 55 | + PREFIX + "_test1", context=context, connection=connection_test |
| 56 | + ) |
| 57 | + schema_empty(Ephys) |
| 58 | + # load the rest of the classes |
| 59 | + schema_empty.spawn_missing_classes(context=context) |
| 60 | + yield schema_empty |
| 61 | + schema_empty.drop() |
| 62 | + |
| 63 | + |
| 64 | +def test_schema_size_on_disk(schema_any): |
| 65 | + number_of_bytes = schema_any.size_on_disk |
| 66 | + assert isinstance(number_of_bytes, int) |
| 67 | + |
| 68 | + |
| 69 | +def test_schema_list(schema_any): |
| 70 | + schemas = dj.list_schemas() |
| 71 | + assert schema_any.database in schemas |
| 72 | + |
| 73 | + |
| 74 | +def test_drop_unauthorized(): |
| 75 | + info_schema = dj.schema("information_schema") |
| 76 | + with pytest.raises(dj.errors.AccessError): |
| 77 | + info_schema.drop() |
| 78 | + |
| 79 | + |
| 80 | +def test_namespace_population(schema_empty_module): |
| 81 | + """ |
| 82 | + With the schema_empty_module fixture, this test |
| 83 | + mimics the behavior of `spawn_missing_classes`, as if the schema |
| 84 | + was declared in a separate module and `spawn_missing_classes` was called in that namespace. |
| 85 | + """ |
| 86 | + # Spawn missing classes in the caller's (self) namespace. |
| 87 | + schema_empty_module.schema.context = None |
| 88 | + schema_empty_module.schema.spawn_missing_classes(context=None) |
| 89 | + # Then add them to the mock module's namespace. |
| 90 | + for k, v in locals().items(): |
| 91 | + if inspect.isclass(v): |
| 92 | + setattr(schema_empty_module, k, v) |
| 93 | + |
| 94 | + for name, rel in getmembers(schema, relation_selector): |
| 95 | + assert hasattr( |
| 96 | + schema_empty_module, name |
| 97 | + ), "{name} not found in schema_empty".format(name=name) |
| 98 | + assert ( |
| 99 | + rel.__base__ is getattr(schema_empty_module, name).__base__ |
| 100 | + ), "Wrong tier for {name}".format(name=name) |
| 101 | + |
| 102 | + for name_part in dir(rel): |
| 103 | + if name_part[0].isupper() and part_selector(getattr(rel, name_part)): |
| 104 | + assert ( |
| 105 | + getattr(rel, name_part).__base__ is dj.Part |
| 106 | + ), "Wrong tier for {name}".format(name=name_part) |
| 107 | + |
| 108 | + |
| 109 | +def test_undecorated_table(): |
| 110 | + """ |
| 111 | + Undecorated user table classes should raise an informative exception upon first use |
| 112 | + """ |
| 113 | + |
| 114 | + class UndecoratedClass(dj.Manual): |
| 115 | + definition = "" |
| 116 | + |
| 117 | + a = UndecoratedClass() |
| 118 | + with pytest.raises(dj.DataJointError): |
| 119 | + print(a.full_table_name) |
| 120 | + |
| 121 | + |
| 122 | +def test_reject_decorated_part(schema_any): |
| 123 | + """ |
| 124 | + Decorating a dj.Part table should raise an informative exception. |
| 125 | + """ |
| 126 | + |
| 127 | + class A(dj.Manual): |
| 128 | + definition = ... |
| 129 | + |
| 130 | + class B(dj.Part): |
| 131 | + definition = ... |
| 132 | + |
| 133 | + with pytest.raises(dj.DataJointError): |
| 134 | + schema_any(A.B) |
| 135 | + schema_any(A) |
| 136 | + |
| 137 | + |
| 138 | +def test_unauthorized_database(db_creds_test): |
| 139 | + """ |
| 140 | + an attempt to create a database to which user has no privileges should raise an informative exception. |
| 141 | + """ |
| 142 | + with pytest.raises(dj.DataJointError): |
| 143 | + dj.Schema( |
| 144 | + "unauthorized_schema", connection=dj.conn(reset=True, **db_creds_test) |
| 145 | + ) |
| 146 | + |
| 147 | + |
| 148 | +def test_drop_database(db_creds_test): |
| 149 | + schema = dj.Schema( |
| 150 | + PREFIX + "_drop_test", connection=dj.conn(reset=True, **db_creds_test) |
| 151 | + ) |
| 152 | + assert schema.exists |
| 153 | + schema.drop() |
| 154 | + assert not schema.exists |
| 155 | + schema.drop() # should do nothing |
| 156 | + |
| 157 | + |
| 158 | +def test_overlapping_name(connection_test): |
| 159 | + test_schema = dj.Schema(PREFIX + "_overlapping_schema", connection=connection_test) |
| 160 | + |
| 161 | + @test_schema |
| 162 | + class Unit(dj.Manual): |
| 163 | + definition = """ |
| 164 | + id: int # simple id |
| 165 | + """ |
| 166 | + |
| 167 | + # hack to update the locals dictionary |
| 168 | + locals() |
| 169 | + |
| 170 | + @test_schema |
| 171 | + class Cell(dj.Manual): |
| 172 | + definition = """ |
| 173 | + type: varchar(32) # type of cell |
| 174 | + """ |
| 175 | + |
| 176 | + class Unit(dj.Part): |
| 177 | + definition = """ |
| 178 | + -> master |
| 179 | + -> Unit |
| 180 | + """ |
| 181 | + |
| 182 | + test_schema.drop() |
| 183 | + |
| 184 | + |
| 185 | +def test_list_tables(schema_simp): |
| 186 | + """ |
| 187 | + https://github.com/datajoint/datajoint-python/issues/838 |
| 188 | + """ |
| 189 | + assert set( |
| 190 | + [ |
| 191 | + "reserved_word", |
| 192 | + "#l", |
| 193 | + "#a", |
| 194 | + "__d", |
| 195 | + "__b", |
| 196 | + "__b__c", |
| 197 | + "__e", |
| 198 | + "__e__f", |
| 199 | + "#outfit_launch", |
| 200 | + "#outfit_launch__outfit_piece", |
| 201 | + "#i_j", |
| 202 | + "#j_i", |
| 203 | + "#t_test_update", |
| 204 | + "#data_a", |
| 205 | + "#data_b", |
| 206 | + "f", |
| 207 | + "#argmax_test", |
| 208 | + "#website", |
| 209 | + "profile", |
| 210 | + "profile__website", |
| 211 | + ] |
| 212 | + ) == set(schema_simp.list_tables()) |
| 213 | + |
| 214 | + |
| 215 | +def test_schema_save_any(schema_any): |
| 216 | + assert "class Experiment(dj.Imported)" in schema_any.code |
| 217 | + |
| 218 | + |
| 219 | +def test_schema_save_empty(schema_empty): |
| 220 | + assert "class Experiment(dj.Imported)" in schema_empty.code |
| 221 | + |
| 222 | + |
| 223 | +def test_uppercase_schema(db_creds_root): |
| 224 | + """ |
| 225 | + https://github.com/datajoint/datajoint-python/issues/564 |
| 226 | + """ |
| 227 | + dj.conn(**db_creds_root, reset=True) |
| 228 | + schema1 = dj.Schema("Schema_A") |
| 229 | + |
| 230 | + @schema1 |
| 231 | + class Subject(dj.Manual): |
| 232 | + definition = """ |
| 233 | + name: varchar(32) |
| 234 | + """ |
| 235 | + |
| 236 | + Schema_A = dj.VirtualModule("Schema_A", "Schema_A") |
| 237 | + |
| 238 | + schema2 = dj.Schema("schema_b") |
| 239 | + |
| 240 | + @schema2 |
| 241 | + class Recording(dj.Manual): |
| 242 | + definition = """ |
| 243 | + -> Schema_A.Subject |
| 244 | + id: smallint |
| 245 | + """ |
| 246 | + |
| 247 | + schema2.drop() |
| 248 | + schema1.drop() |
0 commit comments