Coverage for tests/test_datamodel.py: 99%
773 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-09 08:55 +0000
1# This file is part of felis.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <https://www.gnu.org/licenses/>.
22import difflib
23import os
24import pathlib
25import re
26import shutil
27import tempfile
28import unittest
29from collections import defaultdict
31import yaml
32from lsst.resources import ResourcePath
33from pydantic import ValidationError
35from felis.datamodel import (
36 CheckConstraint,
37 Column,
38 ColumnGroup,
39 ColumnOverrides,
40 Constraint,
41 DataType,
42 ForeignKeyConstraint,
43 Index,
44 Schema,
45 SchemaVersion,
46 Table,
47 UniqueConstraint,
48)
50TEST_DIR = os.path.abspath(os.path.dirname(__file__))
51TEST_YAML = os.path.join(TEST_DIR, "data", "test.yml")
52TEST_SALES = os.path.join(TEST_DIR, "data", "sales.yaml")
53TEST_SERIALIZATION = os.path.join(TEST_DIR, "data", "test_serialization.yaml")
54TEST_ID_GENERATION = os.path.join(TEST_DIR, "data", "test_id_generation.yaml")
57class ColumnTestCase(unittest.TestCase):
58 """Test the ``Column`` class."""
60 def test_validation(self) -> None:
61 """Test Pydantic validation of the ``Column`` class."""
62 # Default initialization should throw an exception.
63 with self.assertRaises(ValidationError):
64 Column()
66 # Setting only name should throw an exception.
67 with self.assertRaises(ValidationError):
68 Column(name="testColumn")
70 # Setting name and id should throw an exception from missing datatype.
71 with self.assertRaises(ValidationError):
72 Column(name="testColumn", id="#test_id")
74 # Setting name, id, and datatype should not throw an exception and
75 # should load data correctly.
76 col = Column(name="testColumn", id="#test_id", datatype="string", length=256)
77 self.assertEqual(col.name, "testColumn", "name should be 'testColumn'")
78 self.assertEqual(col.id, "#test_id", "id should be '#test_id'")
79 self.assertEqual(col.datatype, DataType.string, "datatype should be 'DataType.string'")
81 # Creating from data dictionary should work and load data correctly.
82 data = {"name": "testColumn", "id": "#test_id", "datatype": "string", "length": 256}
83 col = Column(**data)
84 self.assertEqual(col.name, "testColumn", "name should be 'testColumn'")
85 self.assertEqual(col.id, "#test_id", "id should be '#test_id'")
86 self.assertEqual(col.datatype, DataType.string, "datatype should be 'DataType.string'")
88 # Setting a bad IVOA UCD should throw an error.
89 with self.assertRaises(ValidationError):
90 Column(**data, ivoa_ucd="bad")
92 # Setting a valid IVOA UCD should not throw an error.
93 col = Column(**data, ivoa_ucd="meta.id")
94 self.assertEqual(col.ivoa_ucd, "meta.id", "ivoa_ucd should be 'meta.id'")
96 units_data = data.copy()
98 # Setting a bad IVOA unit should throw an error.
99 units_data["ivoa:unit"] = "bad"
100 with self.assertRaises(ValidationError):
101 Column(**units_data)
103 # Setting a valid IVOA unit should not throw an error.
104 units_data["ivoa:unit"] = "m"
105 col = Column(**units_data)
106 self.assertEqual(col.ivoa_unit, "m", "ivoa_unit should be 'm'")
108 units_data = data.copy()
110 # Setting a bad FITS TUNIT should throw an error.
111 units_data["fits:tunit"] = "bad"
112 with self.assertRaises(ValidationError):
113 Column(**units_data)
115 # Setting a valid FITS TUNIT should not throw an error.
116 units_data["fits:tunit"] = "m"
117 col = Column(**units_data)
118 self.assertEqual(col.fits_tunit, "m", "fits_tunit should be 'm'")
120 # Setting both IVOA unit and FITS TUNIT should throw an error.
121 units_data["ivoa:unit"] = "m"
122 with self.assertRaises(ValidationError):
123 Column(**units_data)
125 def test_description_unchecked(self) -> None:
126 """Test that the ``description`` attribute is optional when the
127 ``check_description`` flag is disabled (the default).
128 """
129 # A missing description should be allowed.
130 col = Column.model_validate(
131 {
132 "name": "testColumn",
133 "@id": "#test_col_id",
134 "datatype": "string",
135 "length": 256,
136 }
137 )
138 self.assertIsNone(col.description)
140 # An explicit 'None' description should be allowed.
141 col = Column.model_validate(
142 {
143 "name": "testColumn",
144 "@id": "#test_col_id",
145 "datatype": "string",
146 "length": 256,
147 "description": None,
148 }
149 )
150 self.assertIsNone(col.description)
152 # An empty description should be allowed.
153 col = Column.model_validate(
154 {
155 "name": "testColumn",
156 "@id": "#test_col_id",
157 "datatype": "string",
158 "length": 256,
159 "description": "",
160 }
161 )
162 self.assertEqual(col.description, "")
164 # A description with only whitespace should be allowed.
165 col = Column.model_validate(
166 {
167 "name": "testColumn",
168 "@id": "#test_col_id",
169 "datatype": "string",
170 "length": 256,
171 "description": " ",
172 }
173 )
175 # Pydantic will strip this automatically.
176 self.assertEqual(col.description, "")
178 # A description with one or more non-whitespace characters should be
179 # allowed.
180 col = Column.model_validate(
181 {
182 "name": "testColumn",
183 "@id": "#test_col_id",
184 "datatype": "string",
185 "length": 256,
186 "description": "x",
187 }
188 )
189 self.assertEqual(col.description, "x")
191 def test_description_checked(self) -> None:
192 """Test Pydantic validation of the ``description`` attribute when the
193 ``check_description`` flag is enabled.
194 """
195 cxt = {"check_description": True}
197 # Creating a column with a description of 'None' should throw.
198 with self.assertRaises(ValidationError):
199 Column.model_validate(
200 {
201 "name": "testColumn",
202 "@id": "#test_col_id",
203 "datatype": "string",
204 "length": 256,
205 "description": None,
206 },
207 context=cxt,
208 )
210 # Creating a column with an empty description should throw.
211 with self.assertRaises(ValidationError):
212 Column.model_validate(
213 {
214 "name": "testColumn",
215 "@id": "#test_col_id",
216 "datatype": "string",
217 "length": 256,
218 "description": "",
219 },
220 context=cxt,
221 )
223 # Creating a column with a whitespace-only description should throw,
224 # since whitespace is stripped before validation.
225 with self.assertRaises(ValidationError):
226 Column.model_validate(
227 {
228 "name": "testColumn",
229 "@id": "#test_col_id",
230 "datatype": "string",
231 "length": 256,
232 "description": " ",
233 },
234 context=cxt,
235 )
237 # Creating a column with a single non-whitespace character in the
238 # description should not throw.
239 col = Column.model_validate(
240 {
241 "name": "testColumn",
242 "@id": "#test_col_id",
243 "datatype": "string",
244 "length": 256,
245 "description": "x",
246 },
247 context=cxt,
248 )
249 self.assertEqual(col.description, "x")
251 # Creating a column with more than one non-whitespace character in the
252 # description should not throw.
253 col = Column.model_validate(
254 {
255 "name": "testColumn",
256 "@id": "#test_col_id",
257 "datatype": "string",
258 "length": 256,
259 "description": "test description",
260 },
261 context=cxt,
262 )
263 self.assertEqual(col.description, "test description")
265 def test_values(self) -> None:
266 """Test Pydantic validation of the ``value`` attribute."""
268 # Define a function to return the default column data
269 def default_coldata():
270 return defaultdict(str, {"name": "testColumn", "@id": "#test_col_id"})
272 # Setting both value and autoincrement should throw.
273 autoincr_coldata = default_coldata()
274 autoincr_coldata["datatype"] = "int"
275 autoincr_coldata["autoincrement"] = True
276 autoincr_coldata["value"] = 1
277 with self.assertRaises(ValueError):
278 Column(**autoincr_coldata)
280 # Setting an invalid default on a column with an integer type should
281 # throw.
282 bad_numeric_coldata = default_coldata()
283 for datatype in ["int", "long", "short", "byte"]:
284 for value in ["bad", "1.0", "1", 1.1]:
285 bad_numeric_coldata["datatype"] = datatype
286 bad_numeric_coldata["value"] = value
287 with self.assertRaises(ValueError):
288 Column(**bad_numeric_coldata)
290 # Setting an invalid default on a column with a decimal type should
291 # throw.
292 bad_numeric_coldata = default_coldata()
293 for datatype in ["double", "float"]:
294 for value in ["bad", "1.0", "1", 1]:
295 bad_numeric_coldata["datatype"] = datatype
296 bad_numeric_coldata["value"] = value
297 with self.assertRaises(ValueError):
298 Column(**bad_numeric_coldata)
300 # Setting a bad default on a string column should throw.
301 bad_str_coldata = default_coldata()
302 bad_str_coldata["value"] = 1
303 bad_str_coldata["length"] = 256
304 for datatype in ["string", "char", "unicode", "text"]:
305 for value in [1, 1.1, True, "", " ", " ", "\n", "\t"]:
306 bad_str_coldata["datatype"] = datatype
307 bad_str_coldata["value"] = value
308 with self.assertRaises(ValueError):
309 Column(**bad_str_coldata)
311 # Setting a non-boolean value on a boolean column should throw.
312 bool_coldata = default_coldata()
313 bool_coldata["datatype"] = "boolean"
314 bool_coldata["value"] = "bad"
315 with self.assertRaises(ValueError):
316 for value in ["bad", 1, 1.1]: 316 ↛ 321line 316 didn't jump to line 321
317 bool_coldata["value"] = value
318 Column(**bool_coldata)
320 # Setting a valid value on a string column should be okay.
321 str_coldata = default_coldata()
322 str_coldata["value"] = 1
323 str_coldata["length"] = 256
324 str_coldata["value"] = "okay"
325 for datatype in ["string", "char", "unicode", "text"]:
326 str_coldata["datatype"] = datatype
327 Column(**str_coldata)
329 # Setting an integer value on a column with an int type should be okay.
330 int_coldata = default_coldata()
331 int_coldata["value"] = 1
332 for datatype in ["int", "long", "short", "byte"]:
333 int_coldata["datatype"] = datatype
334 Column(**int_coldata)
336 # Setting a decimal value on a column with a float type should be okay.
337 bool_coldata = default_coldata()
338 bool_coldata["datatype"] = "boolean"
339 bool_coldata["value"] = True
340 Column(**bool_coldata)
342 def test_timestamp(self) -> None:
343 """Test validation of timestamp columns."""
344 # Check that the votable_xtype is set correctly for timestamp columns.
345 col = Column(name="testColumn", id="#test_col_id", datatype="timestamp")
346 self.assertEqual(col.votable_xtype, "timestamp")
349class TableTestCase(unittest.TestCase):
350 """Test Pydantic validation of the ``Table`` class."""
352 def test_validation(self) -> None:
353 """Test Pydantic validation of the ``Table`` class."""
354 # Default initialization should throw an exception.
355 with self.assertRaises(ValidationError):
356 Table()
358 # Setting only name should throw an exception.
359 with self.assertRaises(ValidationError):
360 Table(name="testTable")
362 # Setting name and id should throw an exception from missing columns.
363 with self.assertRaises(ValidationError):
364 Index(name="testTable", id="#test_id")
366 testCol = Column(name="testColumn", id="#test_id", datatype="string", length=256)
368 # Setting name, id, and columns should not throw an exception and
369 # should load data correctly.
370 tbl = Table(name="testTable", id="#test_id", columns=[testCol])
371 self.assertEqual(tbl.name, "testTable", "name should be 'testTable'")
372 self.assertEqual(tbl.id, "#test_id", "id should be '#test_id'")
373 self.assertEqual(tbl.columns, [testCol], "columns should be ['testColumn']")
375 # Creating a table with duplicate column names should raise an
376 # exception.
377 with self.assertRaises(ValidationError):
378 Table(name="testTable", id="#test_id", columns=[testCol, testCol])
381class ColumnGroupTestCase(unittest.TestCase):
382 """Test Pydantic validation of the ``ColumnGroup`` class."""
384 def test_validation(self) -> None:
385 """Test Pydantic validation of the ``ColumnGroup`` class."""
386 # Default initialization should throw an exception.
387 with self.assertRaises(ValidationError):
388 ColumnGroup()
390 # Setting only name should throw an exception.
391 with self.assertRaises(ValidationError):
392 ColumnGroup(name="testGroup")
394 # Setting name and id should throw an exception from missing columns.
395 with self.assertRaises(ValidationError):
396 ColumnGroup(name="testGroup", id="#test_id")
398 col = Column(name="testColumn", id="#test_col", datatype="string", length=256)
400 # Setting name, id, and columns should not throw an exception and
401 # should load data correctly.
402 group = ColumnGroup(name="testGroup", id="#test_group", columns=[col], ivoa_ucd="meta")
403 self.assertEqual(group.name, "testGroup", "name should be 'testGroup'")
404 self.assertEqual(group.id, "#test_group", "id should be '#test_group'")
405 self.assertEqual(group.columns, [col], "columns should be ['testColumn']")
407 # Dereferencing columns without setting a table should raise an
408 # exception.
409 with self.assertRaises(ValueError):
410 group._dereference_columns()
412 # Creating a group with duplicate column names should raise an
413 # exception.
414 with self.assertRaises(ValidationError):
415 ColumnGroup(name="testGroup", id="#test_group", columns=[col, col])
417 # Check that including a column object in a group works correctly.
418 group = ColumnGroup(name="testGroup", id="#test_group", columns=[col], ivoa_ucd="meta")
419 table = Table(
420 name="testTable",
421 id="#test_table",
422 columns=[col],
423 column_groups=[group],
424 )
425 self.assertEqual(table.column_groups, [group], "column_groups should be [group]")
426 self.assertEqual(col, table.column_groups[0].columns[0], "column_groups[0] should be testCol")
428 # Check that column derefencing works correctly when group is assigned
429 # to a table.
430 group = ColumnGroup(name="testGroup", id="#test_group", columns=["#test_col"], ivoa_ucd="meta")
431 table = Table(
432 name="testTable",
433 id="#test_table",
434 columns=[col],
435 column_groups=[group],
436 )
437 self.assertEqual(table.column_groups, [group], "column_groups should be [group]")
438 self.assertEqual(col, table.column_groups[0].columns[0], "column_groups[0] should be testCol")
440 # Creating a group with a bad column should raise an exception.
441 group = ColumnGroup(name="testGroup", id="#test_group", columns=["#bad_col"], ivoa_ucd="meta")
442 with self.assertRaises(ValueError):
443 table = Table(
444 name="testTable",
445 id="#test_table",
446 columns=[col],
447 column_groups=[group],
448 )
451class ConstraintTestCase(unittest.TestCase):
452 """Test Pydantic validation of the different constraint classes."""
454 def test_base_constraint(self) -> None:
455 """Test validation of base constraint type."""
456 # Default initialization should throw an exception.
457 with self.assertRaises(ValidationError):
458 Constraint()
460 # Setting only name should throw an exception.
461 with self.assertRaises(ValidationError):
462 Constraint(name="test_constraint")
464 # Setting name and id should not throw an exception and should load
465 # data correctly.
466 Constraint(name="test_constraint", id="#test_constraint")
468 # Setting initially without deferrable should throw an exception.
469 with self.assertRaises(ValidationError):
470 Constraint(name="test_constraint", id="#test_constraint", deferrable=False, initially="IMMEDIATE")
472 # Seting a bad value for initially should throw an exception.
473 with self.assertRaises(ValidationError):
474 Constraint(name="test_constraint", id="#test_constraint", deferrable=True, initially="BAD_VALUE")
476 # Setting a valid value for initially should not throw an exception.
477 Constraint(name="test_constraint", id="#test_constraint", deferrable=True, initially="IMMEDIATE")
478 Constraint(name="test_constraint", id="#test_constraint", deferrable=True, initially="DEFERRED")
480 def test_unique_constraint(self) -> None:
481 """Test validation of unique constraints."""
482 # Setting name and id should throw an exception from missing columns.
483 with self.assertRaises(ValidationError):
484 UniqueConstraint(name="test_constraint", id="#test_constraint")
486 # Setting name, id, and columns should not throw an exception and
487 # should load data correctly.
488 constraint = UniqueConstraint(name="uniq_test", id="#uniq_test", columns=["test_column"])
489 self.assertEqual(constraint.name, "uniq_test", "name should be 'uniq_test'")
490 self.assertEqual(constraint.id, "#uniq_test", "id should be '#uniq_test'")
491 self.assertEqual(constraint.columns, ["test_column"], "columns should be ['test_column']")
493 # Creating from data dictionary should work and load data correctly.
494 data = {"name": "uniq_test", "id": "#uniq_test", "columns": ["test_column"]}
495 constraint = UniqueConstraint(**data)
496 self.assertEqual(constraint.name, "uniq_test", "name should be 'uniq_test'")
497 self.assertEqual(constraint.id, "#uniq_test", "id should be '#uniq_test'")
498 self.assertEqual(constraint.columns, ["test_column"], "columns should be ['test_column']")
500 def test_foreign_key_constraint(self) -> None:
501 """Test validation of foreign key constraints."""
502 # Setting name and id should throw an exception from missing columns.
503 with self.assertRaises(ValidationError):
504 ForeignKeyConstraint(name="fk_test", id="#fk_test")
506 # Setting name, id, and columns should not throw an exception and
507 # should load data correctly.
508 constraint = ForeignKeyConstraint(
509 name="fk_test", id="#fk_test", columns=["test_column"], referenced_columns=["test_column"]
510 )
511 self.assertEqual(constraint.name, "fk_test", "name should be 'fk_test'")
512 self.assertEqual(constraint.id, "#fk_test", "id should be '#fk_test'")
513 self.assertEqual(constraint.columns, ["test_column"], "columns should be ['test_column']")
514 self.assertEqual(
515 constraint.referenced_columns, ["test_column"], "referenced_columns should be ['test_column']"
516 )
518 # Creating from data dictionary should work and load data correctly.
519 data = {
520 "name": "fk_test",
521 "id": "#fk_test",
522 "columns": ["test_column"],
523 "referenced_columns": ["test_column"],
524 }
525 constraint = ForeignKeyConstraint(**data)
526 self.assertEqual(constraint.name, "fk_test", "name should be 'fk_test'")
527 self.assertEqual(constraint.id, "#fk_test", "id should be '#fk_test'")
528 self.assertEqual(constraint.columns, ["test_column"], "columns should be ['test_column']")
529 self.assertEqual(
530 constraint.referenced_columns, ["test_column"], "referenced_columns should be ['test_column']"
531 )
533 # Creating a foreign key constraint with no columns should raise an
534 # exception.
535 with self.assertRaises(ValidationError):
536 ForeignKeyConstraint(
537 name="fk_test", id="#fk_test", columns=[], referenced_columns=["test_column"]
538 )
540 # Creating a foreign key constraint with no referenced columns should
541 # raise an exception.
542 with self.assertRaises(ValidationError):
543 ForeignKeyConstraint(
544 name="fk_test", id="#fk_test", columns=["test_column"], referenced_columns=[]
545 )
547 # Creating a foreign key constraint where the number of foreign key
548 # columns does not match the number of referenced columns should raise
549 # an exception.
550 with self.assertRaises(ValidationError):
551 ForeignKeyConstraint(
552 name="fk_test",
553 id="#fk_test",
554 columns=["test_column", "test_column2"],
555 referenced_columns=["test_column"],
556 )
558 def test_check_constraint(self) -> None:
559 """Test validation of check constraints."""
560 # Setting name and id should throw an exception from missing
561 # expression.
562 with self.assertRaises(ValidationError):
563 CheckConstraint(name="check_test", id="#check_test")
565 # Setting name, id, and expression should not throw an exception and
566 # should load data correctly.
567 constraint = CheckConstraint(name="check_test", id="#check_test", expression="1+2")
568 self.assertEqual(constraint.name, "check_test", "name should be 'check_test'")
569 self.assertEqual(constraint.id, "#check_test", "id should be '#check_test'")
570 self.assertEqual(constraint.expression, "1+2", "expression should be '1+2'")
572 # Creating from data dictionary should work and load data correctly.
573 data = {
574 "name": "check_test",
575 "id": "#check_test",
576 "expression": "1+2",
577 }
578 constraint = CheckConstraint(**data)
579 self.assertEqual(constraint.name, "check_test", "name should be 'check_test'")
580 self.assertEqual(constraint.id, "#check_test", "id should be '#test_id'")
581 self.assertEqual(constraint.expression, "1+2", "expression should be '1+2'")
583 def test_bad_constraint_type(self) -> None:
584 with self.assertRaises(ValidationError):
585 UniqueConstraint(name="uniq_test", id="#uniq_test", columns=["test_column"], type="BAD_TYPE")
587 def test_constraint_column_checks(self) -> None:
588 """Test the extra validation in the ``Schema`` that checks the
589 constraint column references.
590 """
592 def _create_test_schema(constraint: Constraint) -> None:
593 """Create a test schema with the given constraint."""
594 test_col = Column(name="testColumn", id="#test_col_id", datatype="int")
595 test_col2 = Column(name="testColumn2", id="#test_col_id2", datatype="int")
596 test_tbl = Table(
597 name="testTable", id="#test_tbl_id", columns=[test_col, test_col2], constraints=[constraint]
598 )
599 test_col = Column(name="testColumn", id="#test_col2_id", datatype="int")
600 test_col2 = Column(name="testColumn2", id="#test_col2_id2", datatype="int")
601 test_tbl2 = Table(name="testTable2", id="#test_tbl2_id", columns=[test_col, test_col2])
602 Schema(name="testSchema", id="#test_schema_id", tables=[test_tbl, test_tbl2])
604 # Creating a unique constraint on a bad column should raise an
605 # exception.
606 with self.assertRaises(ValidationError):
607 _create_test_schema(
608 UniqueConstraint(name="testConstraint", id="#test_constraint_id", columns=["bad_column"])
609 )
611 # Creating a foreign key constraint with a bad column should raise an
612 # exception.
613 with self.assertRaises(ValidationError):
614 _create_test_schema(
615 ForeignKeyConstraint(
616 name="testForeignKey",
617 id="#test_fk_id",
618 columns=["bad_column"],
619 referenced_columns=["#test_col2_id"],
620 )
621 )
623 # Creating a foreign key constraint with a bad referenced column should
624 # raise an exception.
625 with self.assertRaises(ValidationError):
626 _create_test_schema(
627 ForeignKeyConstraint(
628 name="testForeignKey",
629 id="#test_fk_id",
630 columns=["#test_col_id"],
631 referenced_columns=["bad_column"],
632 )
633 )
635 # Creating a foreign key constraint where the source column is not in
636 # the same table as the constraint should raise an exception.
637 with self.assertRaises(ValidationError):
638 _create_test_schema(
639 ForeignKeyConstraint(
640 name="testForeignKey",
641 id="#test_fk_id",
642 columns=["#test_col2_id"], # This column is in test_tbl2, not test_tbl
643 referenced_columns=["#test_col_id"],
644 )
645 )
647 # Creating a foreign key constraint where the referenced column is not
648 # a column object should raise an exception.
649 with self.assertRaises(ValidationError):
650 _create_test_schema(
651 ForeignKeyConstraint(
652 name="testForeignKey",
653 id="#test_fk_id",
654 columns=["#test_col_id"],
655 referenced_columns=["#test_schema_id"],
656 )
657 )
659 # Creating a valid unique constraint should not raise an exception.
660 _create_test_schema(
661 UniqueConstraint(name="testConstraint", id="#test_constraint_id", columns=["#test_col_id"])
662 )
664 # Creating a valid foreign key constraint should not raise an
665 # exception.
666 _create_test_schema(
667 ForeignKeyConstraint(
668 name="testForeignKey",
669 id="#test_fk_id",
670 columns=["#test_col_id"],
671 referenced_columns=["#test_col2_id"],
672 )
673 )
675 # Creating a foreign key constraint with a composite key should not
676 # raise an exception.
677 _create_test_schema(
678 ForeignKeyConstraint(
679 name="testCompositeForeignKey",
680 id="#test_composite_fk_id",
681 columns=["#test_col_id", "#test_col_id2"],
682 referenced_columns=["#test_col2_id", "#test_col2_id2"],
683 )
684 )
687class IndexTestCase(unittest.TestCase):
688 """Test Pydantic validation of the ``Index`` class."""
690 def test_index_validation(self) -> None:
691 """Test validation of indexes."""
692 # Default initialization should throw an exception.
693 with self.assertRaises(ValidationError):
694 Index()
696 # Setting only name should throw an exception.
697 with self.assertRaises(ValidationError):
698 Index(name="idx_test")
700 # Setting name and id should throw an exception from missing columns.
701 with self.assertRaises(ValidationError):
702 Index(name="idx_test", id="#idx_test")
704 # Setting name, id, and columns should not throw an exception and
705 # should load data correctly.
706 idx = Index(name="idx_test", id="#idx_test", columns=["#test_column"])
707 self.assertEqual(idx.name, "idx_test", "name should be 'test_constraint'")
708 self.assertEqual(idx.id, "#idx_test", "id should be '#test_id'")
709 self.assertEqual(idx.columns, ["#test_column"], "columns should be ['test_column']")
711 # Creating from data dictionary should work and load data correctly.
712 data = {"name": "idx_test", "id": "#idx_test", "columns": ["test_column"]}
713 idx = Index(**data)
714 self.assertEqual(idx.name, "idx_test", "name should be 'idx_test'")
715 self.assertEqual(idx.id, "#idx_test", "id should be '#idx_test'")
716 self.assertEqual(idx.columns, ["test_column"], "columns should be ['test_column']")
718 # Setting both columns and expressions on an index should throw an
719 # exception.
720 with self.assertRaises(ValidationError):
721 Index(name="idx_test", id="#idx_test", columns=["test_column"], expressions=["1+2"])
724class SchemaTestCase(unittest.TestCase):
725 """Test Pydantic validation of the ``Schema`` class."""
727 def test_validation(self) -> None:
728 """Test Pydantic validation of the main schema class."""
729 # Default initialization should throw an exception.
730 with self.assertRaises(ValidationError):
731 Schema()
733 # Setting only name should throw an exception.
734 with self.assertRaises(ValidationError):
735 Schema(name="testSchema")
737 # Setting name and id should throw an exception from missing columns.
738 with self.assertRaises(ValidationError):
739 Schema(name="testSchema", id="#test_id")
741 test_col = Column(name="testColumn", id="#test_col_id", datatype="string", length=256)
742 test_tbl = Table(name="testTable", id="#test_tbl_id", columns=[test_col])
744 # Setting name, id, and columns should not throw an exception and
745 # should load data correctly.
746 sch = Schema(name="testSchema", id="#test_sch_id", tables=[test_tbl])
747 self.assertEqual(sch.name, "testSchema", "name should be 'testSchema'")
748 self.assertEqual(sch.id, "#test_sch_id", "id should be '#test_sch_id'")
749 self.assertEqual(sch.tables, [test_tbl], "tables should be ['testTable']")
751 # Creating a schema with duplicate table names should raise an
752 # exception.
753 with self.assertRaises(ValidationError):
754 Schema(name="testSchema", id="#test_id", tables=[test_tbl, test_tbl])
756 # Using an undefined YAML field should raise an exception.
757 with self.assertRaises(ValidationError):
758 Schema(**{"name": "testSchema", "id": "#test_sch_id", "bad_field": "1234"}, tables=[test_tbl])
760 # Creating a schema containing duplicate IDs should raise an error.
761 with self.assertRaises(ValidationError):
762 Schema(
763 name="testSchema",
764 id="#test_sch_id",
765 tables=[
766 Table(
767 name="testTable",
768 id="#test_tbl_id",
769 columns=[
770 Column(name="testColumn", id="#test_col_id", datatype="string"),
771 Column(name="testColumn2", id="#test_col_id", datatype="string"),
772 ],
773 )
774 ],
775 )
777 def test_schema_object_ids(self) -> None:
778 """Test that the ``id_map`` is properly populated."""
779 test_col = Column(name="testColumn", id="#test_col_id", datatype="string", length=256)
780 test_tbl = Table(name="testTable", id="#test_table_id", columns=[test_col])
781 sch = Schema(name="testSchema", id="#test_schema_id", tables=[test_tbl])
783 for id in ["#test_col_id", "#test_table_id", "#test_schema_id"]:
784 # Test that the schema contains the expected id.
785 self.assertTrue(id in sch, f"schema should contain '{id}'")
787 # Check that types of returned objects are correct.
788 self.assertIsInstance(sch["#test_col_id"], Column, "schema[id] should return a Column")
789 self.assertIsInstance(sch["#test_table_id"], Table, "schema[id] should return a Table")
790 self.assertIsInstance(sch["#test_schema_id"], Schema, "schema[id] should return a Schema")
792 with self.assertRaises(KeyError):
793 # Test that an invalid id raises an exception.
794 sch["#bad_id"]
796 def test_check_unique_constraint_names(self) -> None:
797 """Test that constraint names are unique."""
798 test_col = Column(name="testColumn", id="#test_col_id", datatype="string", length=256)
799 test_tbl = Table(name="testTable", id="#test_table_id", columns=[test_col])
800 test_cons = UniqueConstraint(name="testConstraint", id="#test_constraint_id", columns=["testColumn"])
801 test_cons2 = UniqueConstraint(
802 name="testConstraint", id="#test_constraint2_id", columns=["testColumn"]
803 )
804 test_tbl.constraints = [test_cons, test_cons2]
805 with self.assertRaises(ValidationError):
806 Schema(name="testSchema", id="#test_id", tables=[test_tbl])
808 def test_check_unique_index_names(self) -> None:
809 """Test that index names are unique."""
810 test_col = Column(name="test_column1", id="#test_table.test_column1", datatype="int")
811 test_col2 = Column(name="test_column2", id="#test_table.test_column2", datatype="string", length=256)
812 test_tbl = Table(name="test_table", id="#test_table", columns=[test_col, test_col2])
813 test_idx = Index(name="idx_test", id="#idx_test", columns=[test_col.id])
814 test_idx2 = Index(name="idx_test", id="#idx_test2", columns=[test_col2.id])
815 test_tbl.indexes = [test_idx, test_idx2]
816 with self.assertRaises(ValidationError):
817 Schema(name="test_schema", id="#test-schema", tables=[test_tbl])
819 def test_model_validate(self) -> None:
820 """Load a YAML test file and validate the schema data model."""
821 with open(TEST_YAML) as test_yaml:
822 data = yaml.safe_load(test_yaml)
823 Schema.model_validate(data)
825 def test_id_generation(self) -> None:
826 """Test ID generation."""
827 test_path = os.path.join(TEST_ID_GENERATION)
828 with open(test_path) as test_yaml:
829 yaml_data = yaml.safe_load(test_yaml)
830 # Generate IDs for objects in the test schema.
831 Schema.model_validate(yaml_data, context={"id_generation": True})
832 with open(test_path) as test_yaml:
833 yaml_data = yaml.safe_load(test_yaml)
834 # Test that an error is raised when id generation is disabled.
835 with self.assertRaises(ValidationError):
836 Schema.model_validate(yaml_data, context={"id_generation": False})
838 def test_get_table_by_column(self) -> None:
839 """Test the ``get_table_by_column`` method."""
840 # Test that the correct table is returned when searching by column.
841 test_col = Column(name="test_column", id="#test_tbl.test_col", datatype="string", length=256)
842 test_tbl = Table(name="test_table", id="#test_tbl", columns=[test_col])
843 sch = Schema(name="testSchema", id="#test_sch_id", tables=[test_tbl])
844 self.assertEqual(sch.get_table_by_column(test_col), test_tbl)
846 # Test that an error is raised when the column is not found.
847 bad_col = Column(name="bad_column", id="#test_tbl.bad_column", datatype="string", length=256)
848 with self.assertRaises(ValueError):
849 sch.get_table_by_column(bad_col)
851 def test_find_object_by_id(self) -> None:
852 test_col = Column(name="test_column", id="#test_tbl.test_col", datatype="string", length=256)
853 test_tbl = Table(name="test_table", id="#test_tbl", columns=[test_col])
854 sch = Schema(name="testSchema", id="#test_sch_id", tables=[test_tbl])
855 self.assertEqual(sch.find_object_by_id("#test_tbl.test_col", Column), test_col)
856 with self.assertRaises(KeyError):
857 sch.find_object_by_id("#bad_id", Column)
858 with self.assertRaises(TypeError):
859 sch.find_object_by_id("#test_tbl", Column)
861 def test_from_file(self) -> None:
862 """Test loading a schema from a file."""
863 # Test file object.
864 with open(TEST_SALES) as test_file:
865 schema = Schema.from_stream(test_file)
866 self.assertIsInstance(schema, Schema)
868 # Test path string.
869 with open(TEST_SALES) as test_file:
870 schema = Schema.from_stream(test_file)
871 self.assertIsInstance(schema, Schema)
873 # Path object.
874 test_file_path = pathlib.Path(TEST_SALES)
875 schema = Schema.from_uri(test_file_path)
876 self.assertIsInstance(schema, Schema)
878 def test_from_resource(self) -> None:
879 """Test loading a schema from a resource."""
880 # Test loading a schema from a resource string.
881 schema = Schema.from_uri(
882 "resource://felis/config/tap_schema/tap_schema_std.yaml", context={"id_generation": True}
883 )
884 self.assertIsInstance(schema, Schema)
886 # Test loading a schema from a ResourcePath.
887 schema = Schema.from_uri(
888 ResourcePath("resource://felis/config/tap_schema/tap_schema_std.yaml"),
889 context={"id_generation": True},
890 )
891 self.assertIsInstance(schema, Schema)
893 # Test loading from a nonexistant resource.
894 with self.assertRaises(ValueError):
895 Schema.from_uri("resource://fake/schemas/bad_schema.yaml")
897 # Without ID generation enabled, this schema should fail validation.
898 with self.assertRaises(ValidationError):
899 Schema.from_uri("resource://felis/config/tap_schema/tap_schema_std.yaml")
901 def test_find_table_by_name(self) -> None:
902 """Test the ``_find_table_by_name`` method."""
903 # Create a simple schema with two tables
904 test_col1 = Column(name="test_column1", id="#test_tbl1.test_col1", datatype="int")
905 test_col2 = Column(name="test_column2", id="#test_tbl2.test_col2", datatype="string", length=256)
906 test_tbl1 = Table(name="test_table1", id="#test_tbl1", columns=[test_col1])
907 test_tbl2 = Table(name="test_table2", id="#test_tbl2", columns=[test_col2])
908 sch = Schema(name="testSchema", id="#test_sch_id", tables=[test_tbl1, test_tbl2])
910 # Test that the correct table is returned when searching by name
911 self.assertEqual(sch._find_table_by_name("test_table1"), test_tbl1)
912 self.assertEqual(sch._find_table_by_name("test_table2"), test_tbl2)
914 # Test that a KeyError is raised when the table is not found
915 with self.assertRaises(KeyError):
916 sch._find_table_by_name("nonexistent_table")
919class SchemaVersionTest(unittest.TestCase):
920 """Test the schema version."""
922 def test_validation(self) -> None:
923 """Test validation of the schema version class."""
924 # Default initialization should throw an exception.
925 with self.assertRaises(ValidationError):
926 SchemaVersion()
928 # Setting current should not throw an exception and should load data
929 # correctly.
930 sv = SchemaVersion(current="1.0.0")
931 self.assertEqual(sv.current, "1.0.0", "current should be '1.0.0'")
933 # Check that schema version can be specified as a single string or
934 # an object.
935 data = {
936 "name": "schema",
937 "@id": "#schema",
938 "tables": [],
939 "version": "1.2.3",
940 }
941 schema = Schema.model_validate(data)
942 self.assertEqual(schema.version, "1.2.3")
944 data = {
945 "name": "schema",
946 "@id": "#schema",
947 "tables": [],
948 "version": {
949 "current": "1.2.3",
950 "compatible": ["1.2.0", "1.2.1", "1.2.2"],
951 "read_compatible": ["1.1.0", "1.1.1"],
952 },
953 }
954 schema = Schema.model_validate(data)
955 self.assertEqual(schema.version.current, "1.2.3")
956 self.assertEqual(schema.version.compatible, ["1.2.0", "1.2.1", "1.2.2"])
957 self.assertEqual(schema.version.read_compatible, ["1.1.0", "1.1.1"])
960class ValidationFlagsTest(unittest.TestCase):
961 """Test optional validation flags on the schema."""
963 def test_check_tap_table_indexes(self) -> None:
964 """Test the ``check_tap_table_indexes`` validation flag."""
965 cxt = {"check_tap_table_indexes": True}
966 schema_dict = {
967 "name": "testSchema",
968 "id": "#test_schema_id",
969 "tables": [
970 {
971 "name": "test_table",
972 "id": "#test_table_id",
973 "columns": [{"name": "test_col", "id": "#test_col", "datatype": "int"}],
974 }
975 ],
976 }
978 # Creating a schema without a TAP table index should throw.
979 with self.assertRaises(ValidationError):
980 Schema.model_validate(schema_dict, context=cxt)
982 # Creating a schema with a TAP table index should not throw.
983 schema_dict["tables"][0]["tap_table_index"] = 1
984 Schema.model_validate(schema_dict, context=cxt)
985 schema_dict["tables"].append(
986 {
987 "name": "test_table2",
988 "id": "#test_table2",
989 "tap_table_index": 1,
990 "columns": [{"name": "test_col2", "id": "#test_col2", "datatype": "int"}],
991 }
992 )
994 # Creating a schema with a duplicate TAP table index should throw.
995 with self.assertRaises(ValidationError):
996 Schema.model_validate(schema_dict, context=cxt)
998 # Multiple, unique TAP table indexes should not throw.
999 schema_dict["tables"][1]["tap_table_index"] = 2
1000 Schema.model_validate(schema_dict, context=cxt)
1002 def test_check_tap_principal(self) -> None:
1003 """Test the ``check_tap_principal` validation flag."""
1004 cxt = {"check_tap_principal": True}
1005 schema_dict = {
1006 "name": "testSchema",
1007 "id": "#test_schema_id",
1008 "tables": [
1009 {
1010 "name": "test_table",
1011 "id": "#test_table_id",
1012 "columns": [{"name": "test_col", "id": "#test_col", "datatype": "int"}],
1013 }
1014 ],
1015 }
1017 # Creating a table without a TAP table principal column should throw.
1018 with self.assertRaises(ValidationError):
1019 Schema.model_validate(schema_dict, context=cxt)
1021 # Creating a table with a TAP table principal column should not throw.
1022 schema_dict["tables"][0]["columns"][0]["tap_principal"] = 1
1023 Schema.model_validate(schema_dict, context=cxt)
1025 def test_check_description(self) -> None:
1026 """Test the ``check_description`` flag."""
1027 cxt = {"check_description": True}
1028 schema_dict = {
1029 "name": "testSchema",
1030 "id": "#test_schema_id",
1031 "tables": [
1032 {
1033 "name": "test_table",
1034 "id": "#test_table_id",
1035 "columns": [{"name": "test_col", "id": "#test_col", "datatype": "int"}],
1036 }
1037 ],
1038 }
1040 # Creating a schema without object descriptions should throw.
1041 with self.assertRaises(ValidationError):
1042 Schema.model_validate(schema_dict, context=cxt)
1044 # Creating a schema with object descriptions should not throw.
1045 schema_dict["description"] = "Test schema"
1046 schema_dict["tables"][0]["description"] = "Test table"
1047 schema_dict["tables"][0]["columns"][0]["description"] = "Test column"
1048 Schema.model_validate(schema_dict, context=cxt)
1051class RedundantDatatypesTest(unittest.TestCase):
1052 """Test validation of redundant datatype definitions."""
1054 def test_mysql_datatypes(self) -> None:
1055 class ColumnGenerator:
1056 """Generate column data for redundant datatype testing."""
1058 def __init__(self, name, id, db_name):
1059 self.name = name
1060 self.id = id
1061 self.db_name = db_name
1062 self.context = {"check_redundant_datatypes": True}
1064 def col(self, datatype: str, db_datatype: str, length=None):
1065 return Column.model_validate(
1066 {
1067 "name": self.name,
1068 "@id": self.id,
1069 "datatype": datatype,
1070 f"{self.db_name}:datatype": db_datatype,
1071 "length": length,
1072 },
1073 context=self.context,
1074 )
1076 """Test that redundant datatype definitions raise an error."""
1077 coldata = ColumnGenerator("test_col", "#test_col_id", "mysql")
1079 with self.assertRaises(ValidationError):
1080 coldata.col("double", "DOUBLE")
1082 with self.assertRaises(ValidationError):
1083 coldata.col("int", "INTEGER")
1085 with self.assertRaises(ValidationError):
1086 coldata.col("float", "FLOAT")
1088 with self.assertRaises(ValidationError):
1089 coldata.col("char", "CHAR", length=8)
1091 with self.assertRaises(ValidationError):
1092 coldata.col("string", "VARCHAR", length=32)
1094 with self.assertRaises(ValidationError):
1095 coldata.col("byte", "TINYINT")
1097 with self.assertRaises(ValidationError):
1098 coldata.col("short", "SMALLINT")
1100 with self.assertRaises(ValidationError):
1101 coldata.col("long", "BIGINT")
1103 with self.assertRaises(ValidationError):
1104 coldata.col("boolean", "BOOLEAN")
1106 with self.assertRaises(ValidationError):
1107 coldata.col("unicode", "NVARCHAR", length=32)
1109 with self.assertRaises(ValidationError):
1110 coldata.col("timestamp", "DATETIME")
1112 # DM-42257: Felis does not handle unbounded text types properly.
1113 # coldata.col("text", "TEXT", length=32)
1115 with self.assertRaises(ValidationError):
1116 coldata.col("binary", "LONGBLOB", length=1024)
1118 with self.assertRaises(ValidationError):
1119 # Same type and length
1120 coldata.col("string", "VARCHAR(128)", length=128)
1122 # Check the old type mapping for MySQL, which is now okay
1123 coldata.col("boolean", "BIT(1)")
1125 # Different types, which is okay
1126 coldata.col("double", "FLOAT")
1128 # Same base type with different lengths, which is okay
1129 coldata.col("string", "VARCHAR(128)", length=32)
1131 # Different string types, which is okay
1132 coldata.col("string", "CHAR", length=32)
1133 coldata.col("unicode", "CHAR", length=32)
1135 def test_precision(self) -> None:
1136 """Test that precision is not allowed for datatypes other than
1137 timestamp.
1138 """
1139 with self.assertRaises(ValidationError):
1140 Column(**{"name": "testColumn", "@id": "#test_col_id", "datatype": "double", "precision": 6})
1143class SchemaSerializationTest(unittest.TestCase):
1144 """Test serialization and deserialization of the schema data model."""
1146 def test_serialization(self) -> None:
1147 """Test serialization of the schema data model."""
1148 # Read the original YAML content from the test_serialization.yaml file
1149 with open(TEST_SERIALIZATION) as file:
1150 original_yaml_content = file.read()
1152 # Load the schema from the original YAML content
1153 schema_out = Schema.from_uri(TEST_SERIALIZATION)
1154 serialized_data = schema_out.model_dump(by_alias=True, exclude_none=True, exclude_defaults=True)
1156 # Write the serialized data to a temporary YAML file
1157 with tempfile.NamedTemporaryFile(delete=False, suffix=".yaml", mode="w+") as temp_file:
1158 yaml.dump(serialized_data, temp_file, default_flow_style=False, sort_keys=False)
1159 temp_file.seek(0)
1160 # Read the deserialized YAML content from the temporary file
1161 deserialized_yaml_content = temp_file.read()
1163 # Show the differences between the original and deserialized YAML
1164 diff = difflib.unified_diff(
1165 original_yaml_content.splitlines(keepends=True),
1166 deserialized_yaml_content.splitlines(keepends=True),
1167 fromfile="original.yaml",
1168 tofile="deserialized.yaml",
1169 )
1170 print("Differences:\n", "".join(diff))
1172 # Assert that the original and deserialized YAML are the same
1173 self.assertEqual(
1174 yaml.safe_load(original_yaml_content),
1175 yaml.safe_load(deserialized_yaml_content),
1176 "The original and deserialized YAML contents should be the same",
1177 )
1180class ResourceTestCase(unittest.TestCase):
1181 """Test loading of column definitions from external schema resources."""
1183 def setUp(self) -> None:
1184 """Set up test resources."""
1185 self.temp_dir = tempfile.mkdtemp()
1187 # Write out source schema file
1188 source_schema_content = """
1189name: source_schema
1190description: Test resource schema
1191tables:
1192- name: source_table
1193 description: Source table
1194 columns:
1195 - name: test_column
1196 datatype: int
1197 description: "Test column"
1198"""
1199 self.source_schema_path = os.path.join(self.temp_dir, "source_schema.yaml")
1200 with open(self.source_schema_path, "w") as f:
1201 f.write(source_schema_content.strip())
1203 # Write out referencing schema file
1204 ref_schema_content = """
1205name: ref_schema
1206description: Test referencing schema
1207resources:
1208 source_schema:
1209 uri: {resource_path}
1210tables:
1211- name: ref_table
1212 description: Referencing table
1213 columnRefs:
1214 source_schema:
1215 source_table:
1216 test_column: null # Explicit null = no overrides, use same name
1217 renamed_column:
1218 ref_name: test_column
1219 overrides:
1220 description: "Renamed test column"
1221 datatype: short
1222 tap:principal: 1
1223 tap:column_index: 2
1224"""
1225 self.ref_schema_path = os.path.join(self.temp_dir, "ref_schema.yaml")
1226 ref_content = ref_schema_content.format(resource_path=self.source_schema_path)
1227 with open(self.ref_schema_path, "w") as f:
1228 f.write(ref_content.strip())
1230 def tearDown(self) -> None:
1231 """Clean up test resources."""
1232 shutil.rmtree(self.temp_dir)
1234 def test_schema_resource(self) -> None:
1235 """Test loading a schema as a resource with column references."""
1236 # First test that the source schema loads correctly on its own
1237 source_schema = Schema.from_uri(self.source_schema_path, context={"id_generation": True})
1238 self.assertEqual(source_schema.name, "source_schema")
1239 self.assertEqual(len(source_schema.tables), 1)
1240 self.assertEqual(source_schema.tables[0].name, "source_table")
1242 # Now test loading the ref schema
1243 ref_schema = Schema.from_uri(self.ref_schema_path, context={"id_generation": True})
1244 self.assertEqual(ref_schema.name, "ref_schema")
1246 # Check that the resource was loaded
1247 self.assertIn("source_schema", ref_schema._resource_map)
1249 # Check that the referencing table has the expected columns
1250 ref_table = ref_schema.tables[0]
1251 self.assertEqual(ref_table.name, "ref_table")
1253 # Check the column_refs structure
1254 column_refs = ref_table.column_refs
1255 self.assertIsNotNone(column_refs)
1256 self.assertIsInstance(column_refs, dict)
1258 # Check the schema resource reference
1259 self.assertIn("source_schema", column_refs)
1260 source_schema_refs = column_refs["source_schema"]
1261 self.assertIsInstance(source_schema_refs, dict)
1263 # Check the table reference
1264 self.assertIn("source_table", source_schema_refs)
1265 source_table_refs = source_schema_refs["source_table"]
1266 self.assertIsInstance(source_table_refs, dict)
1268 # Verify the column_refs structure details
1269 # Should have 2 column references: test_column and renamed_column
1270 self.assertEqual(len(source_table_refs), 2)
1272 # Check test_column reference (null/no overrides)
1273 self.assertIn("test_column", source_table_refs)
1274 test_column_ref = source_table_refs["test_column"]
1275 self.assertIsNone(test_column_ref)
1277 # Check renamed_column reference (with ref_name and overrides)
1278 self.assertIn("renamed_column", source_table_refs)
1279 renamed_column_ref = source_table_refs["renamed_column"]
1280 self.assertIsNotNone(renamed_column_ref)
1281 self.assertEqual(renamed_column_ref.ref_name, "test_column")
1282 self.assertIsNotNone(renamed_column_ref.overrides)
1283 self.assertEqual(renamed_column_ref.overrides.description, "Renamed test column")
1284 self.assertEqual(renamed_column_ref.overrides.tap_principal, 1)
1285 self.assertEqual(renamed_column_ref.overrides.tap_column_index, 2)
1286 self.assertEqual(renamed_column_ref.overrides.datatype.value, "short")
1288 # Now check structure of dereferenced columns in the ref_table
1289 self.assertEqual(len(ref_table.columns), 2)
1291 # Check dereferenced test_column (no overrides)
1292 test_col = next((col for col in ref_table.columns if col.name == "test_column"), None)
1293 self.assertIsNotNone(test_col)
1294 self.assertEqual(test_col.datatype, "int")
1295 self.assertEqual(test_col.description, "Test column")
1297 # Check dereferenced renamed_column (includes overrides)
1298 renamed_col = next((col for col in ref_table.columns if col.name == "renamed_column"), None)
1299 self.assertIsNotNone(renamed_col)
1300 self.assertEqual(renamed_col.datatype, "short") # Inherited from source
1301 self.assertEqual(renamed_col.description, "Renamed test column")
1302 self.assertEqual(renamed_col.tap_principal, 1)
1303 self.assertEqual(renamed_col.tap_column_index, 2)
1305 # Verify that the columns are present in the ID map
1306 try:
1307 ref_schema.find_object_by_id(test_col.id, Column)
1308 except KeyError:
1309 self.fail(f"Test column ID '{test_col.id}' not found in schema ID map.")
1310 try:
1311 ref_schema.find_object_by_id(renamed_col.id, Column)
1312 except KeyError:
1313 self.fail(f"Renamed column ID '{renamed_col.id}' not found in schema ID map.")
1315 def test_schema_resource_missing_column_error(self) -> None:
1316 """Test that referencing a non-existent column raises an error."""
1317 error_ref_content = f"""
1318name: error_ref_schema
1319description: Test referencing schema with error
1320resources:
1321 source_schema:
1322 uri: {self.source_schema_path}
1323tables:
1324- name: error_table
1325 description: Table with bad reference
1326 columnRefs:
1327 source_schema:
1328 source_table:
1329 bad_column: null # This column doesn't exist in source
1330"""
1332 error_ref_path = os.path.join(self.temp_dir, "error_ref_schema.yaml")
1333 with open(error_ref_path, "w") as f:
1334 f.write(error_ref_content.strip())
1336 # This should raise a ValueError
1337 with self.assertRaises(ValueError) as cm:
1338 Schema.from_uri(error_ref_path, context={"id_generation": True})
1340 self.assertIn("Column 'bad_column' not found", str(cm.exception))
1342 def test_schema_resource_missing_ref_name_error(self) -> None:
1343 """Test that using ref_name for non-existent column raises an error."""
1344 # Create a ref schema with bad ref_name
1345 error_ref_content = f"""
1346name: error_ref_schema
1347description: Test referencing schema with bad ref_name
1348resources:
1349 source_schema:
1350 uri: {self.source_schema_path}
1351tables:
1352- name: error_table
1353 description: Table with bad ref_name
1354 columnRefs:
1355 source_schema:
1356 source_table:
1357 some_column:
1358 ref_name: nonexistent_column # This column doesn't exist
1359"""
1361 error_ref_path = os.path.join(self.temp_dir, "error_ref_schema.yaml")
1362 with open(error_ref_path, "w") as f:
1363 f.write(error_ref_content.strip())
1365 with self.assertRaises(ValueError) as cm:
1366 Schema.from_uri(error_ref_path, context={"id_generation": True})
1367 self.assertIn("Column 'nonexistent_column' not found", str(cm.exception))
1369 def test_schema_resource_not_found_error(self) -> None:
1370 """Test that referencing a non-existent schema resource raises an
1371 error.
1372 """
1373 error_ref_content = f"""
1374name: error_ref_schema
1375description: Test referencing non-existent schema resource
1376resources:
1377 source_schema:
1378 uri: {self.source_schema_path}
1379tables:
1380- name: error_table
1381 description: Table with bad schema resource reference
1382 columnRefs:
1383 nonexistent_schema: # This schema resource doesn't exist
1384 some_table:
1385 some_column: null
1386"""
1388 error_ref_path = os.path.join(self.temp_dir, "error_ref_schema.yaml")
1389 with open(error_ref_path, "w") as f:
1390 f.write(error_ref_content.strip())
1392 with self.assertRaises(ValueError) as cm:
1393 Schema.from_uri(error_ref_path, context={"id_generation": True})
1394 self.assertIn("Schema resource 'nonexistent_schema' was not found in resources", str(cm.exception))
1396 def test_schema_resource_table_not_found_error(self) -> None:
1397 """Test that referencing a non-existent table in schema resource raises
1398 an error.
1399 """
1400 error_ref_content = f"""
1401name: error_ref_schema
1402description: Test referencing non-existent table in schema resource
1403resources:
1404 source_schema:
1405 uri: {self.source_schema_path}
1406tables:
1407- name: error_table
1408 description: Table with bad table reference
1409 columnRefs:
1410 source_schema:
1411 nonexistent_table: # This table doesn't exist in source_schema
1412 some_column: null
1413"""
1415 error_ref_path = os.path.join(self.temp_dir, "error_ref_schema.yaml")
1416 with open(error_ref_path, "w") as f:
1417 f.write(error_ref_content.strip())
1419 with self.assertRaises(ValueError) as cm:
1420 Schema.from_uri(error_ref_path, context={"id_generation": True})
1421 self.assertIn("Table 'nonexistent_table' not found in resource 'source_schema'", str(cm.exception))
1423 def test_schema_resource_bad_uri_error(self) -> None:
1424 """Test that a bad URI in resource loading raises an error."""
1425 error_ref_content = """
1426name: error_ref_schema
1427description: Test schema with bad resource URI
1428resources:
1429 bad_resource:
1430 uri: /nonexistent/path/to/schema.yaml
1431tables:
1432- name: error_table
1433 description: Table referencing bad resource
1434 columnRefs:
1435 bad_resource:
1436 some_table:
1437 some_column: null
1438"""
1440 error_ref_path = os.path.join(self.temp_dir, "error_ref_schema.yaml")
1441 with open(error_ref_path, "w") as f:
1442 f.write(error_ref_content.strip())
1444 with self.assertRaises(ValueError) as cm:
1445 Schema.from_uri(error_ref_path, context={"id_generation": True})
1446 self.assertIn(
1447 "Failed to load resource 'bad_resource' from URI 'file:///nonexistent/path/to/schema.yaml'",
1448 str(cm.exception),
1449 )
1451 def test_ref_schema_with_indexes(self) -> None:
1452 """Test that indexes are properly handled when loading schema
1453 resources.
1454 """
1455 # Write out referencing schema file
1456 ref_schema_content_with_indexes = """
1457name: ref_schema
1458description: Test referencing schema
1459resources:
1460 source_schema:
1461 uri: {resource_path}
1462tables:
1463- name: ref_table
1464 description: Referencing table
1465 columnRefs:
1466 source_schema:
1467 source_table:
1468 test_column: null
1469 renamed_column:
1470 ref_name: test_column
1471 overrides:
1472 description: "Renamed test column"
1473 tap:principal: 1
1474 tap:column_index: 2
1475 indexes:
1476 - name: idx_test_column
1477 columns:
1478 - "#ref_table.test_column"
1479 - name: idx_renamed_column
1480 columns:
1481 - "#ref_table.renamed_column"
1482"""
1484 source_schema_with_indexes_path = os.path.join(self.temp_dir, "source_schema_with_indexes.yaml")
1485 ref_content = ref_schema_content_with_indexes.format(resource_path=self.source_schema_path)
1486 with open(source_schema_with_indexes_path, "w") as f:
1487 f.write(ref_content.strip())
1489 ref_schema = Schema.from_uri(source_schema_with_indexes_path, context={"id_generation": True})
1491 # Check index content; columns are not automatically resolved to
1492 # objects by the validation.
1493 indexes = ref_schema.tables[0].indexes
1494 self.assertEqual(len(indexes), 2)
1495 self.assertEqual(indexes[0].name, "idx_test_column")
1496 self.assertEqual(indexes[0].columns, ["#ref_table.test_column"])
1497 self.assertEqual(indexes[1].name, "idx_renamed_column")
1498 self.assertEqual(indexes[1].columns, ["#ref_table.renamed_column"])
1500 def test_ref_schema_with_foreign_key(self) -> None:
1501 """Test that indexes are properly handled when loading schema
1502 resources.
1503 """
1504 # Write out referencing schema file
1505 ref_schema_content_with_foreign_key = """
1506name: ref_schema
1507description: Test referencing schema
1508resources:
1509 source_schema:
1510 uri: {resource_path}
1511tables:
1512- name: src_table
1513 description: Source table for foreign key
1514 primaryKey: "#src_table.test_column"
1515 columnRefs:
1516 source_schema:
1517 source_table:
1518 test_column: null
1519 renamed_column:
1520 ref_name: test_column
1521 overrides:
1522 description: "Renamed test column"
1523 tap:principal: 1
1524 tap:column_index: 2
1525- name: target_table
1526 description: Target table for foreign key
1527 columns:
1528 - name: fk_column
1529 datatype: int
1530 description: "Foreign key column"
1531 constraints:
1532 - name: fk_src_table
1533 '@type': ForeignKey
1534 columns:
1535 - "#target_table.fk_column"
1536 referencedColumns:
1537 - "#src_table.test_column"
1538"""
1540 source_schema_with_foreign_key_path = os.path.join(
1541 self.temp_dir, "source_schema_with_foreign_key.yaml"
1542 )
1543 ref_content = ref_schema_content_with_foreign_key.format(resource_path=self.source_schema_path)
1544 with open(source_schema_with_foreign_key_path, "w") as f:
1545 f.write(ref_content.strip())
1547 ref_schema = Schema.from_uri(source_schema_with_foreign_key_path, context={"id_generation": True})
1549 # Check foreign key constraint content
1550 fk_constraint = ref_schema.tables[1].constraints[0]
1551 self.assertIsInstance(fk_constraint, ForeignKeyConstraint)
1552 self.assertEqual(fk_constraint.name, "fk_src_table")
1553 self.assertEqual(fk_constraint.columns, ["#target_table.fk_column"])
1554 self.assertEqual(fk_constraint.referenced_columns, ["#src_table.test_column"])
1556 def test_ref_schema_serialization(self) -> None:
1557 """Test serialization of a reference schema."""
1558 # Load the referencing schema and then serialize it back to YAML
1559 ref_schema = Schema.from_uri(self.ref_schema_path, context={"id_generation": True})
1560 yaml_data = ref_schema.model_dump(by_alias=True, exclude_none=True, exclude_defaults=True)
1561 serialized_schema_path = os.path.join(self.temp_dir, "serialized_ref_schema.yaml")
1562 with open(serialized_schema_path, "w") as f:
1563 yaml.dump(yaml_data, f, default_flow_style=False, sort_keys=False)
1565 # Read back the serialized YAML data
1566 with open(serialized_schema_path) as f:
1567 serialized_yaml_data = yaml.safe_load(f)
1569 # Ensure that columns were not serialized directly in the table
1570 self.assertEqual(len(serialized_yaml_data["tables"][0]["columns"]), 0)
1572 # Deserialize the schema and check that the expected columns are
1573 # present
1574 deserialized_schema = Schema.from_uri(serialized_schema_path, context={"id_generation": True})
1575 self.assertEqual(len(deserialized_schema.tables[0].columns), 2)
1577 # Check that the columnRefs structure is still present
1578 try:
1579 ref_columns = deserialized_schema.tables[0].column_refs["source_schema"]["source_table"]
1580 except Exception:
1581 self.fail("The column refs are missing after deserialization.")
1582 self.assertEqual(len(ref_columns), 2)
1584 def test_ref_schema_with_dereference_columns(self) -> None:
1585 """Test loading a reference schema with dereferencing of columns so
1586 that column_refs is set to empty after loading.
1587 """
1588 ref_schema = Schema.from_uri(
1589 self.ref_schema_path, context={"id_generation": True, "dereference_resources": True}
1590 )
1592 # Check that the columns were dereferenced into the table
1593 ref_table = ref_schema.tables[0]
1594 self.assertEqual(len(ref_table.columns), 2)
1595 col_names = {col.name for col in ref_table.columns}
1596 self.assertIn("test_column", col_names)
1597 self.assertIn("renamed_column", col_names)
1599 # Check that column_refs is empty after dereferencing
1600 self.assertEqual(len(ref_table.column_refs), 0)
1602 def test_tap_column_index_with_overrides(self) -> None:
1603 """Test that TAP column index is correctly assigned when an override
1604 of that field is present in the column ref.
1605 """
1606 # Write out source schema file
1607 source_schema_content = """
1608name: source_schema
1609tables:
1610- name: source_table
1611 columns:
1612 - name: col1
1613 datatype: int
1614 - name: col2
1615 datatype: int
1616 - name: col3
1617 datatype: int
1618"""
1619 source_schema_path = os.path.join(self.temp_dir, "source_schema.yaml")
1620 with open(source_schema_path, "w") as f:
1621 f.write(source_schema_content.strip())
1623 # Write out referencing schema file
1624 ref_schema_content = """
1625name: ref_schema
1626resources:
1627 source_schema:
1628 uri: {resource_path}
1629tables:
1630- name: ref_table
1631 columnRefs:
1632 source_schema:
1633 source_table:
1634 col1:
1635 col2:
1636 overrides:
1637 tap:column_index: 15
1638 col3:
1639"""
1640 ref_schema_path = os.path.join(self.temp_dir, "ref_schema.yaml")
1641 ref_content = ref_schema_content.format(resource_path=source_schema_path)
1642 with open(ref_schema_path, "w") as f:
1643 f.write(ref_content.strip())
1645 ref_schema = Schema.from_uri(
1646 ref_schema_path,
1647 context={"id_generation": True, "column_ref_index_increment": 10},
1648 )
1650 for column in ref_schema.tables[0].columns:
1651 if column.name == "col1":
1652 self.assertEqual(column.tap_column_index, 10)
1653 elif column.name == "col2":
1654 self.assertEqual(column.tap_column_index, 15)
1655 elif column.name == "col3": 1655 ↛ 1658line 1655 didn't jump to line 1658 because the condition on line 1655 was always true
1656 self.assertEqual(column.tap_column_index, 20)
1657 else:
1658 self.fail(f"Unexpected column name: {column.name}")
1660 def test_from_uri_with_relative_resource_uri(self) -> None:
1661 """Test that from_uri resolves a relative resource URI using the
1662 schema's resource_path context.
1663 """
1664 source_content = """
1665name: source_schema
1666tables:
1667- name: source_table
1668 columns:
1669 - name: test_column
1670 datatype: int
1671"""
1672 source_path = os.path.join(self.temp_dir, "source_schema.yaml")
1673 with open(source_path, "w") as f:
1674 f.write(source_content.strip())
1676 # Write referencing schema using a relative URI.
1677 ref_content = """
1678name: ref_schema
1679resources:
1680 source_schema:
1681 uri: source_schema.yaml
1682tables:
1683- name: ref_table
1684 columnRefs:
1685 source_schema:
1686 source_table:
1687 test_column:
1688"""
1689 ref_path = os.path.join(self.temp_dir, "ref_schema.yaml")
1690 with open(ref_path, "w") as f:
1691 f.write(ref_content.strip())
1693 base_rp = ResourcePath(ref_path)
1694 resolved_rp = base_rp.parent().join("source_schema.yaml")
1695 self.assertEqual(resolved_rp.ospath, source_path)
1697 # Load via from_uri; the relative resource URI should be resolved
1698 # against the referencing schema's directory.
1699 schema = Schema.from_uri(ref_path, context={"id_generation": True})
1701 # Verify the resource was found and loaded without errors
1702 self.assertIn("source_schema", schema._resource_map)
1704 def test_from_uri_with_absolute_resource_uri(self) -> None:
1705 """Test that from_uri works with an absolute resource URI."""
1706 source_content = """
1707name: source_schema
1708tables:
1709- name: source_table
1710 columns:
1711 - name: test_column
1712 datatype: int
1713"""
1714 source_path = os.path.join(self.temp_dir, "source_schema.yaml")
1715 with open(source_path, "w") as f:
1716 f.write(source_content.strip())
1718 # Write referencing schema that uses an absolute URI for the resource
1719 ref_content = f"""
1720name: ref_schema
1721resources:
1722 source_schema:
1723 uri: {source_path}
1724tables:
1725- name: ref_table
1726 columnRefs:
1727 source_schema:
1728 source_table:
1729 test_column:
1730"""
1731 ref_path = os.path.join(self.temp_dir, "ref_schema.yaml")
1732 with open(ref_path, "w") as f:
1733 f.write(ref_content.strip())
1735 # Verify the expected resolved path from joining absolute URI.
1736 from lsst.resources import ResourcePath
1738 base_rp = ResourcePath(ref_path)
1739 resolved_rp = base_rp.join(source_path)
1740 self.assertEqual(resolved_rp.ospath, source_path)
1742 # Load via from_uri with an absolute resource URI.
1743 schema = Schema.from_uri(ref_path, context={"id_generation": True})
1745 # Verify the resource was found and loaded without errors.
1746 self.assertIn("source_schema", schema._resource_map)
1748 def test_absolute_resource_uri_ignores_ref_schema_directory(self) -> None:
1749 """Test that an absolute resource URI is not resolved relative to
1750 the referencing schema's directory.
1751 """
1752 # Put the source schema in its own subdirectory.
1753 source_dir = os.path.join(self.temp_dir, "source_dir")
1754 os.makedirs(source_dir)
1756 source_content = """
1757name: source_schema
1758tables:
1759- name: source_table
1760 columns:
1761 - name: test_column
1762 datatype: int
1763"""
1764 source_path = os.path.join(source_dir, "source_schema.yaml")
1765 with open(source_path, "w") as f:
1766 f.write(source_content.strip())
1768 # Put the referencing schema in a different subdirectory.
1769 ref_dir = os.path.join(self.temp_dir, "ref_dir")
1770 os.makedirs(ref_dir)
1772 ref_content = f"""
1773name: ref_schema
1774resources:
1775 source_schema:
1776 uri: {source_path}
1777tables:
1778- name: ref_table
1779 columnRefs:
1780 source_schema:
1781 source_table:
1782 test_column:
1783"""
1784 ref_path = os.path.join(ref_dir, "ref_schema.yaml")
1785 with open(ref_path, "w") as f:
1786 f.write(ref_content.strip())
1788 # Load via from_uri. The absolute URI should resolve directly,
1789 # ignoring the referencing schema's directory.
1790 schema = Schema.from_uri(ref_path, context={"id_generation": True})
1792 # Verify the resource was found and loaded without errors.
1793 self.assertIn("source_schema", schema._resource_map)
1796class ColumnOverridesTestCase(unittest.TestCase):
1797 """Test application of overrides to a column, setting all allowed
1798 fields.
1799 """
1801 def test_all_override_fields_exist_on_column(self) -> None:
1802 """Ensure every ColumnOverrides field corresponds to an attribute on
1803 Column.
1804 """
1805 override_fields = set(ColumnOverrides.model_fields)
1806 column_fields = set(Column.model_fields)
1808 missing = override_fields - column_fields
1810 self.assertFalse(
1811 missing,
1812 f"Column is missing attributes for override fields: {sorted(missing)}",
1813 )
1815 def test_overrides_all(self) -> None:
1816 """Test updating all allowed column fields from overrides."""
1817 # Create a base column
1818 base_column = Column(
1819 name="base_column",
1820 id="#base_column",
1821 description="Base column",
1822 datatype="char",
1823 length=64,
1824 nullable=False,
1825 tap_principal=1,
1826 tap_column_index=10,
1827 )
1829 # Override all allowed fields with different values
1830 overrides = ColumnOverrides(
1831 description="Ref column",
1832 datatype="string",
1833 length=256,
1834 nullable=True,
1835 tap_principal=0,
1836 tap_column_index=100,
1837 )
1839 # Apply overrides
1840 base_column._update_from_overrides(overrides)
1842 # Check that the attributes were updated correctly
1843 self.assertEqual(base_column.description, "Ref column")
1844 self.assertEqual(base_column.datatype, "string")
1845 self.assertEqual(base_column.length, 256)
1846 self.assertEqual(base_column.nullable, True)
1847 self.assertEqual(base_column.tap_principal, 0)
1848 self.assertEqual(base_column.tap_column_index, 100)
1850 def test_overrides_subset(self) -> None:
1851 """Test updating a subset of allowed column fields from overrides."""
1852 # Create a base column
1853 base_column = Column(
1854 name="base_column",
1855 id="#base_column",
1856 description="Base column",
1857 datatype="char",
1858 length=64,
1859 nullable=False,
1860 tap_principal=1,
1861 tap_column_index=10,
1862 )
1864 # Override all allowed fields with different values
1865 overrides = ColumnOverrides(
1866 description="Ref column",
1867 tap_column_index=100,
1868 )
1870 # Apply overrides
1871 base_column._update_from_overrides(overrides)
1873 # Check that the attributes were updated correctly
1874 self.assertEqual(base_column.description, "Ref column")
1875 self.assertEqual(base_column.datatype, "char")
1876 self.assertEqual(base_column.length, 64)
1877 self.assertEqual(base_column.nullable, False)
1878 self.assertEqual(base_column.tap_principal, 1)
1879 self.assertEqual(base_column.tap_column_index, 100)
1881 def test_overrides_default(self) -> None:
1882 """Test that applying the default overrides is a no-op."""
1883 # Create a base column
1884 base_column = Column(
1885 name="base_column",
1886 id="#base_column",
1887 description="Base column",
1888 datatype="char",
1889 length=64,
1890 nullable=False,
1891 tap_principal=1,
1892 tap_column_index=10,
1893 )
1895 # Apply overrides
1896 base_column._update_from_overrides(ColumnOverrides())
1898 # Check that the attributes remain unchanged
1899 self.assertEqual(base_column.description, "Base column")
1900 self.assertEqual(base_column.datatype, "char")
1901 self.assertEqual(base_column.length, 64)
1902 self.assertEqual(base_column.nullable, False)
1903 self.assertEqual(base_column.tap_principal, 1)
1904 self.assertEqual(base_column.tap_column_index, 10)
1906 def test_overrides_with_explicit_none_values(self) -> None:
1907 """Test that passing explicit None values in overrides does update
1908 the column attributes where allowed and raises errors if it is not.
1909 """
1910 # Create a base column
1911 base_column = Column(
1912 name="base_column",
1913 id="#base_column",
1914 description="Base column",
1915 datatype="int",
1916 length=64,
1917 nullable=False,
1918 tap_principal=1,
1919 tap_column_index=10,
1920 )
1922 # Create overrides with explicit None values for nullable fields
1923 overrides = ColumnOverrides(
1924 description=None,
1925 tap_column_index=None,
1926 )
1928 # Apply overrides
1929 base_column._update_from_overrides(overrides)
1931 # Check that the attributes were updated to None where allowed
1932 self.assertIsNone(base_column.description)
1933 self.assertIsNone(base_column.tap_column_index)
1935 # Check that setting non-nullable fields to None raise a specific
1936 # ValueError on ColumnOverrides creation
1937 for non_nullable_field in ("datatype", "length", "nullable", "tap_principal"):
1938 with self.assertRaisesRegex(
1939 ValueError,
1940 re.escape(f"The '{non_nullable_field}' field cannot be overridden to null"),
1941 ):
1942 ColumnOverrides(**{non_nullable_field: None})
1944 def test_extra_fields_in_overrides(self) -> None:
1945 """Test that extra fields in ColumnOverrides raise a
1946 ValidationError.
1947 """
1948 with self.assertRaises(ValidationError) as cm:
1949 ColumnOverrides(
1950 description="Test column",
1951 extra_field="This should not be allowed",
1952 )
1954 self.assertIn("Extra inputs are not permitted", str(cm.exception))
1956 def test_overrides_accept_alias_keys(self) -> None:
1957 """Test that alias keys for TAP fields are accepted and populate the
1958 corresponding model fields.
1959 """
1960 overrides = ColumnOverrides(**{"tap:principal": 1, "tap:column_index": 42})
1962 self.assertEqual(overrides.tap_principal, 1)
1963 self.assertEqual(overrides.tap_column_index, 42)
1965 # Ensure these count as explicitly provided (for model_fields_set
1966 # logic).
1967 self.assertIn("tap_principal", overrides.model_fields_set)
1968 self.assertIn("tap_column_index", overrides.model_fields_set)
1970 def test_datatype_deserialize_and_serialize(self) -> None:
1971 """Test that datatype is deserialized from a string to DataType and
1972 serialized back to a string.
1973 """
1974 overrides = ColumnOverrides(datatype="char")
1976 # Deserialization should yield a DataType instance (not a raw str).
1977 self.assertIsInstance(overrides.datatype, DataType)
1978 self.assertEqual(str(overrides.datatype), "char")
1980 # Serialization should produce a JSON-friendly string value.
1981 dumped = overrides.model_dump(mode="json")
1982 self.assertEqual(dumped["datatype"], "char")
1984 # None should remain None on serialization.
1985 overrides_none = ColumnOverrides()
1986 dumped_none = overrides_none.model_dump(mode="json")
1987 self.assertIsNone(dumped_none["datatype"])
1989 def test_non_nullable_overrides_data_is_none(self) -> None:
1990 """Test that passing None to ``_check_non_nullable_overrides`` does not
1991 raise an error.
1992 """
1993 ColumnOverrides()._check_non_nullable_overrides(None)
1996if __name__ == "__main__": 1996 ↛ 1997line 1996 didn't jump to line 1997 because the condition on line 1996 was never true
1997 unittest.main()