Coverage for tests/test_datamodel.py: 11%

1# This file is part of felis.

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.

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 shutil

26import tempfile

27import unittest

28from collections import defaultdict

30import yaml

31from lsst.resources import ResourcePath

32from pydantic import ValidationError

34from felis.datamodel import (

35 CheckConstraint,

36 Column,

37 ColumnGroup,

38 Constraint,

39 DataType,

40 ForeignKeyConstraint,

41 Index,

42 Schema,

43 SchemaVersion,

44 Table,

45 UniqueConstraint,

46)

48TEST_DIR = os.path.abspath(os.path.dirname(__file__))

49TEST_YAML = os.path.join(TEST_DIR, "data", "test.yml")

50TEST_SALES = os.path.join(TEST_DIR, "data", "sales.yaml")

51TEST_SERIALIZATION = os.path.join(TEST_DIR, "data", "test_serialization.yaml")

52TEST_ID_GENERATION = os.path.join(TEST_DIR, "data", "test_id_generation.yaml")

55class ColumnTestCase(unittest.TestCase):

56 """Test the ``Column`` class."""

58 def test_validation(self) -> None:

59 """Test Pydantic validation of the ``Column`` class."""

60 # Default initialization should throw an exception.

61 with self.assertRaises(ValidationError):

62 Column()

64 # Setting only name should throw an exception.

65 with self.assertRaises(ValidationError):

66 Column(name="testColumn")

68 # Setting name and id should throw an exception from missing datatype.

69 with self.assertRaises(ValidationError):

70 Column(name="testColumn", id="#test_id")

72 # Setting name, id, and datatype should not throw an exception and

73 # should load data correctly.

74 col = Column(name="testColumn", id="#test_id", datatype="string", length=256)

75 self.assertEqual(col.name, "testColumn", "name should be 'testColumn'")

76 self.assertEqual(col.id, "#test_id", "id should be '#test_id'")

77 self.assertEqual(col.datatype, DataType.string, "datatype should be 'DataType.string'")

79 # Creating from data dictionary should work and load data correctly.

80 data = {"name": "testColumn", "id": "#test_id", "datatype": "string", "length": 256}

81 col = Column(**data)

82 self.assertEqual(col.name, "testColumn", "name should be 'testColumn'")

83 self.assertEqual(col.id, "#test_id", "id should be '#test_id'")

84 self.assertEqual(col.datatype, DataType.string, "datatype should be 'DataType.string'")

86 # Setting a bad IVOA UCD should throw an error.

87 with self.assertRaises(ValidationError):

88 Column(**data, ivoa_ucd="bad")

90 # Setting a valid IVOA UCD should not throw an error.

91 col = Column(**data, ivoa_ucd="meta.id")

92 self.assertEqual(col.ivoa_ucd, "meta.id", "ivoa_ucd should be 'meta.id'")

94 units_data = data.copy()

96 # Setting a bad IVOA unit should throw an error.

97 units_data["ivoa:unit"] = "bad"

98 with self.assertRaises(ValidationError):

99 Column(**units_data)

100

101 # Setting a valid IVOA unit should not throw an error.

102 units_data["ivoa:unit"] = "m"

103 col = Column(**units_data)

104 self.assertEqual(col.ivoa_unit, "m", "ivoa_unit should be 'm'")

105

106 units_data = data.copy()

107

108 # Setting a bad FITS TUNIT should throw an error.

109 units_data["fits:tunit"] = "bad"

110 with self.assertRaises(ValidationError):

111 Column(**units_data)

112

113 # Setting a valid FITS TUNIT should not throw an error.

114 units_data["fits:tunit"] = "m"

115 col = Column(**units_data)

116 self.assertEqual(col.fits_tunit, "m", "fits_tunit should be 'm'")

117

118 # Setting both IVOA unit and FITS TUNIT should throw an error.

119 units_data["ivoa:unit"] = "m"

120 with self.assertRaises(ValidationError):

121 Column(**units_data)

122

123 def test_description(self) -> None:

124 """Test Pydantic validation of the ``description`` attribute."""

125 # Creating a column with a description of 'None' should throw.

126 with self.assertRaises(ValueError):

127 Column(

128 **{

129 "name": "testColumn",

130 "@id": "#test_col_id",

131 "datatype": "string",

132 "description": None,

133 }

134 )

135

136 # Creating a column with an empty description should throw.

137 with self.assertRaises(ValueError):

138 Column(

139 **{

140 "name": "testColumn",

141 "@id": "#test_col_id",

142 "datatype": "string",

143 "description": "",

144 }

145 )

146

147 # Creating a column with a description that is too short should throw.

148 with self.assertRaises(ValidationError):

149 Column(

150 **{

151 "name": "testColumn",

152 "@id": "#test_col_id",

153 "datatype": "string",

154 "description": "xy",

155 }

156 )

157

158 def test_values(self) -> None:

159 """Test Pydantic validation of the ``value`` attribute."""

160

161 # Define a function to return the default column data

162 def default_coldata():

163 return defaultdict(str, {"name": "testColumn", "@id": "#test_col_id"})

164

165 # Setting both value and autoincrement should throw.

166 autoincr_coldata = default_coldata()

167 autoincr_coldata["datatype"] = "int"

168 autoincr_coldata["autoincrement"] = True

169 autoincr_coldata["value"] = 1

170 with self.assertRaises(ValueError):

171 Column(**autoincr_coldata)

172

173 # Setting an invalid default on a column with an integer type should

174 # throw.

175 bad_numeric_coldata = default_coldata()

176 for datatype in ["int", "long", "short", "byte"]:

177 for value in ["bad", "1.0", "1", 1.1]:

178 bad_numeric_coldata["datatype"] = datatype

179 bad_numeric_coldata["value"] = value

180 with self.assertRaises(ValueError):

181 Column(**bad_numeric_coldata)

182

183 # Setting an invalid default on a column with a decimal type should

184 # throw.

185 bad_numeric_coldata = default_coldata()

186 for datatype in ["double", "float"]:

187 for value in ["bad", "1.0", "1", 1]:

188 bad_numeric_coldata["datatype"] = datatype

189 bad_numeric_coldata["value"] = value

190 with self.assertRaises(ValueError):

191 Column(**bad_numeric_coldata)

192

193 # Setting a bad default on a string column should throw.

194 bad_str_coldata = default_coldata()

195 bad_str_coldata["value"] = 1

196 bad_str_coldata["length"] = 256

197 for datatype in ["string", "char", "unicode", "text"]:

198 for value in [1, 1.1, True, "", " ", " ", "\n", "\t"]:

199 bad_str_coldata["datatype"] = datatype

200 bad_str_coldata["value"] = value

201 with self.assertRaises(ValueError):

202 Column(**bad_str_coldata)

203

204 # Setting a non-boolean value on a boolean column should throw.

205 bool_coldata = default_coldata()

206 bool_coldata["datatype"] = "boolean"

207 bool_coldata["value"] = "bad"

208 with self.assertRaises(ValueError):

209 for value in ["bad", 1, 1.1]:

210 bool_coldata["value"] = value

211 Column(**bool_coldata)

212

213 # Setting a valid value on a string column should be okay.

214 str_coldata = default_coldata()

215 str_coldata["value"] = 1

216 str_coldata["length"] = 256

217 str_coldata["value"] = "okay"

218 for datatype in ["string", "char", "unicode", "text"]:

219 str_coldata["datatype"] = datatype

220 Column(**str_coldata)

221

222 # Setting an integer value on a column with an int type should be okay.

223 int_coldata = default_coldata()

224 int_coldata["value"] = 1

225 for datatype in ["int", "long", "short", "byte"]:

226 int_coldata["datatype"] = datatype

227 Column(**int_coldata)

228

229 # Setting a decimal value on a column with a float type should be okay.

230 bool_coldata = default_coldata()

231 bool_coldata["datatype"] = "boolean"

232 bool_coldata["value"] = True

233 Column(**bool_coldata)

234

235 def test_timestamp(self) -> None:

236 """Test validation of timestamp columns."""

237 # Check that the votable_xtype is set correctly for timestamp columns.

238 col = Column(name="testColumn", id="#test_col_id", datatype="timestamp")

239 self.assertEqual(col.votable_xtype, "timestamp")

240

241

242class TableTestCase(unittest.TestCase):

243 """Test Pydantic validation of the ``Table`` class."""

244

245 def test_validation(self) -> None:

246 """Test Pydantic validation of the ``Table`` class."""

247 # Default initialization should throw an exception.

248 with self.assertRaises(ValidationError):

249 Table()

250

251 # Setting only name should throw an exception.

252 with self.assertRaises(ValidationError):

253 Table(name="testTable")

254

255 # Setting name and id should throw an exception from missing columns.

256 with self.assertRaises(ValidationError):

257 Index(name="testTable", id="#test_id")

258

259 testCol = Column(name="testColumn", id="#test_id", datatype="string", length=256)

260

261 # Setting name, id, and columns should not throw an exception and

262 # should load data correctly.

263 tbl = Table(name="testTable", id="#test_id", columns=[testCol])

264 self.assertEqual(tbl.name, "testTable", "name should be 'testTable'")

265 self.assertEqual(tbl.id, "#test_id", "id should be '#test_id'")

266 self.assertEqual(tbl.columns, [testCol], "columns should be ['testColumn']")

267

268 # Creating a table with duplicate column names should raise an

269 # exception.

270 with self.assertRaises(ValidationError):

271 Table(name="testTable", id="#test_id", columns=[testCol, testCol])

272

273

274class ColumnGroupTestCase(unittest.TestCase):

275 """Test Pydantic validation of the ``ColumnGroup`` class."""

276

277 def test_validation(self) -> None:

278 """Test Pydantic validation of the ``ColumnGroup`` class."""

279 # Default initialization should throw an exception.

280 with self.assertRaises(ValidationError):

281 ColumnGroup()

282

283 # Setting only name should throw an exception.

284 with self.assertRaises(ValidationError):

285 ColumnGroup(name="testGroup")

286

287 # Setting name and id should throw an exception from missing columns.

288 with self.assertRaises(ValidationError):

289 ColumnGroup(name="testGroup", id="#test_id")

290

291 col = Column(name="testColumn", id="#test_col", datatype="string", length=256)

292

293 # Setting name, id, and columns should not throw an exception and

294 # should load data correctly.

295 group = ColumnGroup(name="testGroup", id="#test_group", columns=[col], ivoa_ucd="meta")

296 self.assertEqual(group.name, "testGroup", "name should be 'testGroup'")

297 self.assertEqual(group.id, "#test_group", "id should be '#test_group'")

298 self.assertEqual(group.columns, [col], "columns should be ['testColumn']")

299

300 # Dereferencing columns without setting a table should raise an

301 # exception.

302 with self.assertRaises(ValueError):

303 group._dereference_columns()

304

305 # Creating a group with duplicate column names should raise an

306 # exception.

307 with self.assertRaises(ValidationError):

308 ColumnGroup(name="testGroup", id="#test_group", columns=[col, col])

309

310 # Check that including a column object in a group works correctly.

311 group = ColumnGroup(name="testGroup", id="#test_group", columns=[col], ivoa_ucd="meta")

312 table = Table(

313 name="testTable",

314 id="#test_table",

315 columns=[col],

316 column_groups=[group],

317 )

318 self.assertEqual(table.column_groups, [group], "column_groups should be [group]")

319 self.assertEqual(col, table.column_groups[0].columns[0], "column_groups[0] should be testCol")

320

321 # Check that column derefencing works correctly when group is assigned

322 # to a table.

323 group = ColumnGroup(name="testGroup", id="#test_group", columns=["#test_col"], ivoa_ucd="meta")

324 table = Table(

325 name="testTable",

326 id="#test_table",

327 columns=[col],

328 column_groups=[group],

329 )

330 self.assertEqual(table.column_groups, [group], "column_groups should be [group]")

331 self.assertEqual(col, table.column_groups[0].columns[0], "column_groups[0] should be testCol")

332

333 # Creating a group with a bad column should raise an exception.

334 group = ColumnGroup(name="testGroup", id="#test_group", columns=["#bad_col"], ivoa_ucd="meta")

335 with self.assertRaises(ValueError):

336 table = Table(

337 name="testTable",

338 id="#test_table",

339 columns=[col],

340 column_groups=[group],

341 )

342

343

344class ConstraintTestCase(unittest.TestCase):

345 """Test Pydantic validation of the different constraint classes."""

346

347 def test_base_constraint(self) -> None:

348 """Test validation of base constraint type."""

349 # Default initialization should throw an exception.

350 with self.assertRaises(ValidationError):

351 Constraint()

352

353 # Setting only name should throw an exception.

354 with self.assertRaises(ValidationError):

355 Constraint(name="test_constraint")

356

357 # Setting name and id should not throw an exception and should load

358 # data correctly.

359 Constraint(name="test_constraint", id="#test_constraint")

360

361 # Setting initially without deferrable should throw an exception.

362 with self.assertRaises(ValidationError):

363 Constraint(name="test_constraint", id="#test_constraint", deferrable=False, initially="IMMEDIATE")

364

365 # Seting a bad value for initially should throw an exception.

366 with self.assertRaises(ValidationError):

367 Constraint(name="test_constraint", id="#test_constraint", deferrable=True, initially="BAD_VALUE")

368

369 # Setting a valid value for initially should not throw an exception.

370 Constraint(name="test_constraint", id="#test_constraint", deferrable=True, initially="IMMEDIATE")

371 Constraint(name="test_constraint", id="#test_constraint", deferrable=True, initially="DEFERRED")

372

373 def test_unique_constraint(self) -> None:

374 """Test validation of unique constraints."""

375 # Setting name and id should throw an exception from missing columns.

376 with self.assertRaises(ValidationError):

377 UniqueConstraint(name="test_constraint", id="#test_constraint")

378

379 # Setting name, id, and columns should not throw an exception and

380 # should load data correctly.

381 constraint = UniqueConstraint(name="uniq_test", id="#uniq_test", columns=["test_column"])

382 self.assertEqual(constraint.name, "uniq_test", "name should be 'uniq_test'")

383 self.assertEqual(constraint.id, "#uniq_test", "id should be '#uniq_test'")

384 self.assertEqual(constraint.columns, ["test_column"], "columns should be ['test_column']")

385

386 # Creating from data dictionary should work and load data correctly.

387 data = {"name": "uniq_test", "id": "#uniq_test", "columns": ["test_column"]}

388 constraint = UniqueConstraint(**data)

389 self.assertEqual(constraint.name, "uniq_test", "name should be 'uniq_test'")

390 self.assertEqual(constraint.id, "#uniq_test", "id should be '#uniq_test'")

391 self.assertEqual(constraint.columns, ["test_column"], "columns should be ['test_column']")

392

393 def test_foreign_key_constraint(self) -> None:

394 """Test validation of foreign key constraints."""

395 # Setting name and id should throw an exception from missing columns.

396 with self.assertRaises(ValidationError):

397 ForeignKeyConstraint(name="fk_test", id="#fk_test")

398

399 # Setting name, id, and columns should not throw an exception and

400 # should load data correctly.

401 constraint = ForeignKeyConstraint(

402 name="fk_test", id="#fk_test", columns=["test_column"], referenced_columns=["test_column"]

403 )

404 self.assertEqual(constraint.name, "fk_test", "name should be 'fk_test'")

405 self.assertEqual(constraint.id, "#fk_test", "id should be '#fk_test'")

406 self.assertEqual(constraint.columns, ["test_column"], "columns should be ['test_column']")

407 self.assertEqual(

408 constraint.referenced_columns, ["test_column"], "referenced_columns should be ['test_column']"

409 )

410

411 # Creating from data dictionary should work and load data correctly.

412 data = {

413 "name": "fk_test",

414 "id": "#fk_test",

415 "columns": ["test_column"],

416 "referenced_columns": ["test_column"],

417 }

418 constraint = ForeignKeyConstraint(**data)

419 self.assertEqual(constraint.name, "fk_test", "name should be 'fk_test'")

420 self.assertEqual(constraint.id, "#fk_test", "id should be '#fk_test'")

421 self.assertEqual(constraint.columns, ["test_column"], "columns should be ['test_column']")

422 self.assertEqual(

423 constraint.referenced_columns, ["test_column"], "referenced_columns should be ['test_column']"

424 )

425

426 # Creating a foreign key constraint with no columns should raise an

427 # exception.

428 with self.assertRaises(ValidationError):

429 ForeignKeyConstraint(

430 name="fk_test", id="#fk_test", columns=[], referenced_columns=["test_column"]

431 )

432

433 # Creating a foreign key constraint with no referenced columns should

434 # raise an exception.

435 with self.assertRaises(ValidationError):

436 ForeignKeyConstraint(

437 name="fk_test", id="#fk_test", columns=["test_column"], referenced_columns=[]

438 )

439

440 # Creating a foreign key constraint where the number of foreign key

441 # columns does not match the number of referenced columns should raise

442 # an exception.

443 with self.assertRaises(ValidationError):

444 ForeignKeyConstraint(

445 name="fk_test",

446 id="#fk_test",

447 columns=["test_column", "test_column2"],

448 referenced_columns=["test_column"],

449 )

450

451 def test_check_constraint(self) -> None:

452 """Test validation of check constraints."""

453 # Setting name and id should throw an exception from missing

454 # expression.

455 with self.assertRaises(ValidationError):

456 CheckConstraint(name="check_test", id="#check_test")

457

458 # Setting name, id, and expression should not throw an exception and

459 # should load data correctly.

460 constraint = CheckConstraint(name="check_test", id="#check_test", expression="1+2")

461 self.assertEqual(constraint.name, "check_test", "name should be 'check_test'")

462 self.assertEqual(constraint.id, "#check_test", "id should be '#check_test'")

463 self.assertEqual(constraint.expression, "1+2", "expression should be '1+2'")

464

465 # Creating from data dictionary should work and load data correctly.

466 data = {

467 "name": "check_test",

468 "id": "#check_test",

469 "expression": "1+2",

470 }

471 constraint = CheckConstraint(**data)

472 self.assertEqual(constraint.name, "check_test", "name should be 'check_test'")

473 self.assertEqual(constraint.id, "#check_test", "id should be '#test_id'")

474 self.assertEqual(constraint.expression, "1+2", "expression should be '1+2'")

475

476 def test_bad_constraint_type(self) -> None:

477 with self.assertRaises(ValidationError):

478 UniqueConstraint(name="uniq_test", id="#uniq_test", columns=["test_column"], type="BAD_TYPE")

479

480 def test_constraint_column_checks(self) -> None:

481 """Test the extra validation in the ``Schema`` that checks the

482 constraint column references.

483 """

484

485 def _create_test_schema(constraint: Constraint) -> None:

486 """Create a test schema with the given constraint."""

487 test_col = Column(name="testColumn", id="#test_col_id", datatype="int")

488 test_col2 = Column(name="testColumn2", id="#test_col_id2", datatype="int")

489 test_tbl = Table(

490 name="testTable", id="#test_tbl_id", columns=[test_col, test_col2], constraints=[constraint]

491 )

492 test_col = Column(name="testColumn", id="#test_col2_id", datatype="int")

493 test_col2 = Column(name="testColumn2", id="#test_col2_id2", datatype="int")

494 test_tbl2 = Table(name="testTable2", id="#test_tbl2_id", columns=[test_col, test_col2])

495 Schema(name="testSchema", id="#test_schema_id", tables=[test_tbl, test_tbl2])

496

497 # Creating a unique constraint on a bad column should raise an

498 # exception.

499 with self.assertRaises(ValidationError):

500 _create_test_schema(

501 UniqueConstraint(name="testConstraint", id="#test_constraint_id", columns=["bad_column"])

502 )

503

504 # Creating a foreign key constraint with a bad column should raise an

505 # exception.

506 with self.assertRaises(ValidationError):

507 _create_test_schema(

508 ForeignKeyConstraint(

509 name="testForeignKey",

510 id="#test_fk_id",

511 columns=["bad_column"],

512 referenced_columns=["#test_col2_id"],

513 )

514 )

515

516 # Creating a foreign key constraint with a bad referenced column should

517 # raise an exception.

518 with self.assertRaises(ValidationError):

519 _create_test_schema(

520 ForeignKeyConstraint(

521 name="testForeignKey",

522 id="#test_fk_id",

523 columns=["#test_col_id"],

524 referenced_columns=["bad_column"],

525 )

526 )

527

528 # Creating a foreign key constraint where the source column is not in

529 # the same table as the constraint should raise an exception.

530 with self.assertRaises(ValidationError):

531 _create_test_schema(

532 ForeignKeyConstraint(

533 name="testForeignKey",

534 id="#test_fk_id",

535 columns=["#test_col2_id"], # This column is in test_tbl2, not test_tbl

536 referenced_columns=["#test_col_id"],

537 )

538 )

539

540 # Creating a foreign key constraint where the referenced column is not

541 # a column object should raise an exception.

542 with self.assertRaises(ValidationError):

543 _create_test_schema(

544 ForeignKeyConstraint(

545 name="testForeignKey",

546 id="#test_fk_id",

547 columns=["#test_col_id"],

548 referenced_columns=["#test_schema_id"],

549 )

550 )

551

552 # Creating a valid unique constraint should not raise an exception.

553 _create_test_schema(

554 UniqueConstraint(name="testConstraint", id="#test_constraint_id", columns=["#test_col_id"])

555 )

556

557 # Creating a valid foreign key constraint should not raise an

558 # exception.

559 _create_test_schema(

560 ForeignKeyConstraint(

561 name="testForeignKey",

562 id="#test_fk_id",

563 columns=["#test_col_id"],

564 referenced_columns=["#test_col2_id"],

565 )

566 )

567

568 # Creating a foreign key constraint with a composite key should not

569 # raise an exception.

570 _create_test_schema(

571 ForeignKeyConstraint(

572 name="testCompositeForeignKey",

573 id="#test_composite_fk_id",

574 columns=["#test_col_id", "#test_col_id2"],

575 referenced_columns=["#test_col2_id", "#test_col2_id2"],

576 )

577 )

578

579

580class IndexTestCase(unittest.TestCase):

581 """Test Pydantic validation of the ``Index`` class."""

582

583 def test_index_validation(self) -> None:

584 """Test validation of indexes."""

585 # Default initialization should throw an exception.

586 with self.assertRaises(ValidationError):

587 Index()

588

589 # Setting only name should throw an exception.

590 with self.assertRaises(ValidationError):

591 Index(name="idx_test")

592

593 # Setting name and id should throw an exception from missing columns.

594 with self.assertRaises(ValidationError):

595 Index(name="idx_test", id="#idx_test")

596

597 # Setting name, id, and columns should not throw an exception and

598 # should load data correctly.

599 idx = Index(name="idx_test", id="#idx_test", columns=["#test_column"])

600 self.assertEqual(idx.name, "idx_test", "name should be 'test_constraint'")

601 self.assertEqual(idx.id, "#idx_test", "id should be '#test_id'")

602 self.assertEqual(idx.columns, ["#test_column"], "columns should be ['test_column']")

603

604 # Creating from data dictionary should work and load data correctly.

605 data = {"name": "idx_test", "id": "#idx_test", "columns": ["test_column"]}

606 idx = Index(**data)

607 self.assertEqual(idx.name, "idx_test", "name should be 'idx_test'")

608 self.assertEqual(idx.id, "#idx_test", "id should be '#idx_test'")

609 self.assertEqual(idx.columns, ["test_column"], "columns should be ['test_column']")

610

611 # Setting both columns and expressions on an index should throw an

612 # exception.

613 with self.assertRaises(ValidationError):

614 Index(name="idx_test", id="#idx_test", columns=["test_column"], expressions=["1+2"])

615

616

617class SchemaTestCase(unittest.TestCase):

618 """Test Pydantic validation of the ``Schema`` class."""

619

620 def test_validation(self) -> None:

621 """Test Pydantic validation of the main schema class."""

622 # Default initialization should throw an exception.

623 with self.assertRaises(ValidationError):

624 Schema()

625

626 # Setting only name should throw an exception.

627 with self.assertRaises(ValidationError):

628 Schema(name="testSchema")

629

630 # Setting name and id should throw an exception from missing columns.

631 with self.assertRaises(ValidationError):

632 Schema(name="testSchema", id="#test_id")

633

634 test_col = Column(name="testColumn", id="#test_col_id", datatype="string", length=256)

635 test_tbl = Table(name="testTable", id="#test_tbl_id", columns=[test_col])

636

637 # Setting name, id, and columns should not throw an exception and

638 # should load data correctly.

639 sch = Schema(name="testSchema", id="#test_sch_id", tables=[test_tbl])

640 self.assertEqual(sch.name, "testSchema", "name should be 'testSchema'")

641 self.assertEqual(sch.id, "#test_sch_id", "id should be '#test_sch_id'")

642 self.assertEqual(sch.tables, [test_tbl], "tables should be ['testTable']")

643

644 # Creating a schema with duplicate table names should raise an

645 # exception.

646 with self.assertRaises(ValidationError):

647 Schema(name="testSchema", id="#test_id", tables=[test_tbl, test_tbl])

648

649 # Using an undefined YAML field should raise an exception.

650 with self.assertRaises(ValidationError):

651 Schema(**{"name": "testSchema", "id": "#test_sch_id", "bad_field": "1234"}, tables=[test_tbl])

652

653 # Creating a schema containing duplicate IDs should raise an error.

654 with self.assertRaises(ValidationError):

655 Schema(

656 name="testSchema",

657 id="#test_sch_id",

658 tables=[

659 Table(

660 name="testTable",

661 id="#test_tbl_id",

662 columns=[

663 Column(name="testColumn", id="#test_col_id", datatype="string"),

664 Column(name="testColumn2", id="#test_col_id", datatype="string"),

665 ],

666 )

667 ],

668 )

669

670 def test_schema_object_ids(self) -> None:

671 """Test that the ``id_map`` is properly populated."""

672 test_col = Column(name="testColumn", id="#test_col_id", datatype="string", length=256)

673 test_tbl = Table(name="testTable", id="#test_table_id", columns=[test_col])

674 sch = Schema(name="testSchema", id="#test_schema_id", tables=[test_tbl])

675

676 for id in ["#test_col_id", "#test_table_id", "#test_schema_id"]:

677 # Test that the schema contains the expected id.

678 self.assertTrue(id in sch, f"schema should contain '{id}'")

679

680 # Check that types of returned objects are correct.

681 self.assertIsInstance(sch["#test_col_id"], Column, "schema[id] should return a Column")

682 self.assertIsInstance(sch["#test_table_id"], Table, "schema[id] should return a Table")

683 self.assertIsInstance(sch["#test_schema_id"], Schema, "schema[id] should return a Schema")

684

685 with self.assertRaises(KeyError):

686 # Test that an invalid id raises an exception.

687 sch["#bad_id"]

688

689 def test_check_unique_constraint_names(self) -> None:

690 """Test that constraint names are unique."""

691 test_col = Column(name="testColumn", id="#test_col_id", datatype="string", length=256)

692 test_tbl = Table(name="testTable", id="#test_table_id", columns=[test_col])

693 test_cons = UniqueConstraint(name="testConstraint", id="#test_constraint_id", columns=["testColumn"])

694 test_cons2 = UniqueConstraint(

695 name="testConstraint", id="#test_constraint2_id", columns=["testColumn"]

696 )

697 test_tbl.constraints = [test_cons, test_cons2]

698 with self.assertRaises(ValidationError):

699 Schema(name="testSchema", id="#test_id", tables=[test_tbl])

700

701 def test_check_unique_index_names(self) -> None:

702 """Test that index names are unique."""

703 test_col = Column(name="test_column1", id="#test_table.test_column1", datatype="int")

704 test_col2 = Column(name="test_column2", id="#test_table.test_column2", datatype="string", length=256)

705 test_tbl = Table(name="test_table", id="#test_table", columns=[test_col, test_col2])

706 test_idx = Index(name="idx_test", id="#idx_test", columns=[test_col.id])

707 test_idx2 = Index(name="idx_test", id="#idx_test2", columns=[test_col2.id])

708 test_tbl.indexes = [test_idx, test_idx2]

709 with self.assertRaises(ValidationError):

710 Schema(name="test_schema", id="#test-schema", tables=[test_tbl])

711

712 def test_model_validate(self) -> None:

713 """Load a YAML test file and validate the schema data model."""

714 with open(TEST_YAML) as test_yaml:

715 data = yaml.safe_load(test_yaml)

716 Schema.model_validate(data)

717

718 def test_id_generation(self) -> None:

719 """Test ID generation."""

720 test_path = os.path.join(TEST_ID_GENERATION)

721 with open(test_path) as test_yaml:

722 yaml_data = yaml.safe_load(test_yaml)

723 # Generate IDs for objects in the test schema.

724 Schema.model_validate(yaml_data, context={"id_generation": True})

725 with open(test_path) as test_yaml:

726 yaml_data = yaml.safe_load(test_yaml)

727 # Test that an error is raised when id generation is disabled.

728 with self.assertRaises(ValidationError):

729 Schema.model_validate(yaml_data, context={"id_generation": False})

730

731 def test_get_table_by_column(self) -> None:

732 """Test the ``get_table_by_column`` method."""

733 # Test that the correct table is returned when searching by column.

734 test_col = Column(name="test_column", id="#test_tbl.test_col", datatype="string", length=256)

735 test_tbl = Table(name="test_table", id="#test_tbl", columns=[test_col])

736 sch = Schema(name="testSchema", id="#test_sch_id", tables=[test_tbl])

737 self.assertEqual(sch.get_table_by_column(test_col), test_tbl)

738

739 # Test that an error is raised when the column is not found.

740 bad_col = Column(name="bad_column", id="#test_tbl.bad_column", datatype="string", length=256)

741 with self.assertRaises(ValueError):

742 sch.get_table_by_column(bad_col)

743

744 def test_find_object_by_id(self) -> None:

745 test_col = Column(name="test_column", id="#test_tbl.test_col", datatype="string", length=256)

746 test_tbl = Table(name="test_table", id="#test_tbl", columns=[test_col])

747 sch = Schema(name="testSchema", id="#test_sch_id", tables=[test_tbl])

748 self.assertEqual(sch.find_object_by_id("#test_tbl.test_col", Column), test_col)

749 with self.assertRaises(KeyError):

750 sch.find_object_by_id("#bad_id", Column)

751 with self.assertRaises(TypeError):

752 sch.find_object_by_id("#test_tbl", Column)

753

754 def test_from_file(self) -> None:

755 """Test loading a schema from a file."""

756 # Test file object.

757 with open(TEST_SALES) as test_file:

758 schema = Schema.from_stream(test_file)

759 self.assertIsInstance(schema, Schema)

760

761 # Test path string.

762 with open(TEST_SALES) as test_file:

763 schema = Schema.from_stream(test_file)

764 self.assertIsInstance(schema, Schema)

765

766 # Path object.

767 test_file_path = pathlib.Path(TEST_SALES)

768 schema = Schema.from_uri(test_file_path)

769 self.assertIsInstance(schema, Schema)

770

771 def test_from_resource(self) -> None:

772 """Test loading a schema from a resource."""

773 # Test loading a schema from a resource string.

774 schema = Schema.from_uri(

775 "resource://felis/config/tap_schema/tap_schema_std.yaml", context={"id_generation": True}

776 )

777 self.assertIsInstance(schema, Schema)

778

779 # Test loading a schema from a ResourcePath.

780 schema = Schema.from_uri(

781 ResourcePath("resource://felis/config/tap_schema/tap_schema_std.yaml"),

782 context={"id_generation": True},

783 )

784 self.assertIsInstance(schema, Schema)

785

786 # Test loading from a nonexistant resource.

787 with self.assertRaises(ValueError):

788 Schema.from_uri("resource://fake/schemas/bad_schema.yaml")

789

790 # Without ID generation enabled, this schema should fail validation.

791 with self.assertRaises(ValidationError):

792 Schema.from_uri("resource://felis/config/tap_schema/tap_schema_std.yaml")

793

794 def test_find_table_by_name(self) -> None:

795 """Test the ``_find_table_by_name`` method."""

796 # Create a simple schema with two tables

797 test_col1 = Column(name="test_column1", id="#test_tbl1.test_col1", datatype="int")

798 test_col2 = Column(name="test_column2", id="#test_tbl2.test_col2", datatype="string", length=256)

799 test_tbl1 = Table(name="test_table1", id="#test_tbl1", columns=[test_col1])

800 test_tbl2 = Table(name="test_table2", id="#test_tbl2", columns=[test_col2])

801 sch = Schema(name="testSchema", id="#test_sch_id", tables=[test_tbl1, test_tbl2])

802

803 # Test that the correct table is returned when searching by name

804 self.assertEqual(sch._find_table_by_name("test_table1"), test_tbl1)

805 self.assertEqual(sch._find_table_by_name("test_table2"), test_tbl2)

806

807 # Test that a KeyError is raised when the table is not found

808 with self.assertRaises(KeyError):

809 sch._find_table_by_name("nonexistent_table")

810

811

812class SchemaVersionTest(unittest.TestCase):

813 """Test the schema version."""

814

815 def test_validation(self) -> None:

816 """Test validation of the schema version class."""

817 # Default initialization should throw an exception.

818 with self.assertRaises(ValidationError):

819 SchemaVersion()

820

821 # Setting current should not throw an exception and should load data

822 # correctly.

823 sv = SchemaVersion(current="1.0.0")

824 self.assertEqual(sv.current, "1.0.0", "current should be '1.0.0'")

825

826 # Check that schema version can be specified as a single string or

827 # an object.

828 data = {

829 "name": "schema",

830 "@id": "#schema",

831 "tables": [],

832 "version": "1.2.3",

833 }

834 schema = Schema.model_validate(data)

835 self.assertEqual(schema.version, "1.2.3")

836

837 data = {

838 "name": "schema",

839 "@id": "#schema",

840 "tables": [],

841 "version": {

842 "current": "1.2.3",

843 "compatible": ["1.2.0", "1.2.1", "1.2.2"],

844 "read_compatible": ["1.1.0", "1.1.1"],

845 },

846 }

847 schema = Schema.model_validate(data)

848 self.assertEqual(schema.version.current, "1.2.3")

849 self.assertEqual(schema.version.compatible, ["1.2.0", "1.2.1", "1.2.2"])

850 self.assertEqual(schema.version.read_compatible, ["1.1.0", "1.1.1"])

851

852

853class ValidationFlagsTest(unittest.TestCase):

854 """Test optional validation flags on the schema."""

855

856 def test_check_tap_table_indexes(self) -> None:

857 """Test the ``check_tap_table_indexes`` validation flag."""

858 cxt = {"check_tap_table_indexes": True}

859 schema_dict = {

860 "name": "testSchema",

861 "id": "#test_schema_id",

862 "tables": [

863 {

864 "name": "test_table",

865 "id": "#test_table_id",

866 "columns": [{"name": "test_col", "id": "#test_col", "datatype": "int"}],

867 }

868 ],

869 }

870

871 # Creating a schema without a TAP table index should throw.

872 with self.assertRaises(ValidationError):

873 Schema.model_validate(schema_dict, context=cxt)

874

875 # Creating a schema with a TAP table index should not throw.

876 schema_dict["tables"][0]["tap_table_index"] = 1

877 Schema.model_validate(schema_dict, context=cxt)

878 schema_dict["tables"].append(

879 {

880 "name": "test_table2",

881 "id": "#test_table2",

882 "tap_table_index": 1,

883 "columns": [{"name": "test_col2", "id": "#test_col2", "datatype": "int"}],

884 }

885 )

886

887 # Creating a schema with a duplicate TAP table index should throw.

888 with self.assertRaises(ValidationError):

889 Schema.model_validate(schema_dict, context=cxt)

890

891 # Multiple, unique TAP table indexes should not throw.

892 schema_dict["tables"][1]["tap_table_index"] = 2

893 Schema.model_validate(schema_dict, context=cxt)

894

895 def test_check_tap_principal(self) -> None:

896 """Test the ``check_tap_principal` validation flag."""

897 cxt = {"check_tap_principal": True}

898 schema_dict = {

899 "name": "testSchema",

900 "id": "#test_schema_id",

901 "tables": [

902 {

903 "name": "test_table",

904 "id": "#test_table_id",

905 "columns": [{"name": "test_col", "id": "#test_col", "datatype": "int"}],

906 }

907 ],

908 }

909

910 # Creating a table without a TAP table principal column should throw.

911 with self.assertRaises(ValidationError):

912 Schema.model_validate(schema_dict, context=cxt)

913

914 # Creating a table with a TAP table principal column should not throw.

915 schema_dict["tables"][0]["columns"][0]["tap_principal"] = 1

916 Schema.model_validate(schema_dict, context=cxt)

917

918 def test_check_description(self) -> None:

919 """Test the ``check_description`` flag."""

920 cxt = {"check_description": True}

921 schema_dict = {

922 "name": "testSchema",

923 "id": "#test_schema_id",

924 "tables": [

925 {

926 "name": "test_table",

927 "id": "#test_table_id",

928 "columns": [{"name": "test_col", "id": "#test_col", "datatype": "int"}],

929 }

930 ],

931 }

932

933 # Creating a schema without object descriptions should throw.

934 with self.assertRaises(ValidationError):

935 Schema.model_validate(schema_dict, context=cxt)

936

937 # Creating a schema with object descriptions should not throw.

938 schema_dict["description"] = "Test schema"

939 schema_dict["tables"][0]["description"] = "Test table"

940 schema_dict["tables"][0]["columns"][0]["description"] = "Test column"

941 Schema.model_validate(schema_dict, context=cxt)

942

943

944class RedundantDatatypesTest(unittest.TestCase):

945 """Test validation of redundant datatype definitions."""

946

947 def test_mysql_datatypes(self) -> None:

948 class ColumnGenerator:

949 """Generate column data for redundant datatype testing."""

950

951 def __init__(self, name, id, db_name):

952 self.name = name

953 self.id = id

954 self.db_name = db_name

955 self.context = {"check_redundant_datatypes": True}

956

957 def col(self, datatype: str, db_datatype: str, length=None):

958 return Column.model_validate(

959 {

960 "name": self.name,

961 "@id": self.id,

962 "datatype": datatype,

963 f"{self.db_name}:datatype": db_datatype,

964 "length": length,

965 },

966 context=self.context,

967 )

968

969 """Test that redundant datatype definitions raise an error."""

970 coldata = ColumnGenerator("test_col", "#test_col_id", "mysql")

971

972 with self.assertRaises(ValidationError):

973 coldata.col("double", "DOUBLE")

974

975 with self.assertRaises(ValidationError):

976 coldata.col("int", "INTEGER")

977

978 with self.assertRaises(ValidationError):

979 coldata.col("float", "FLOAT")

980

981 with self.assertRaises(ValidationError):

982 coldata.col("char", "CHAR", length=8)

983

984 with self.assertRaises(ValidationError):

985 coldata.col("string", "VARCHAR", length=32)

986

987 with self.assertRaises(ValidationError):

988 coldata.col("byte", "TINYINT")

989

990 with self.assertRaises(ValidationError):

991 coldata.col("short", "SMALLINT")

992

993 with self.assertRaises(ValidationError):

994 coldata.col("long", "BIGINT")

995

996 with self.assertRaises(ValidationError):

997 coldata.col("boolean", "BOOLEAN")

998

999 with self.assertRaises(ValidationError):

1000 coldata.col("unicode", "NVARCHAR", length=32)

1001

1002 with self.assertRaises(ValidationError):

1003 coldata.col("timestamp", "DATETIME")

1004

1005 # DM-42257: Felis does not handle unbounded text types properly.

1006 # coldata.col("text", "TEXT", length=32)

1007

1008 with self.assertRaises(ValidationError):

1009 coldata.col("binary", "LONGBLOB", length=1024)

1010

1011 with self.assertRaises(ValidationError):

1012 # Same type and length

1013 coldata.col("string", "VARCHAR(128)", length=128)

1014

1015 # Check the old type mapping for MySQL, which is now okay

1016 coldata.col("boolean", "BIT(1)")

1017

1018 # Different types, which is okay

1019 coldata.col("double", "FLOAT")

1020

1021 # Same base type with different lengths, which is okay

1022 coldata.col("string", "VARCHAR(128)", length=32)

1023

1024 # Different string types, which is okay

1025 coldata.col("string", "CHAR", length=32)

1026 coldata.col("unicode", "CHAR", length=32)

1027

1028 def test_precision(self) -> None:

1029 """Test that precision is not allowed for datatypes other than

1030 timestamp.

1031 """

1032 with self.assertRaises(ValidationError):

1033 Column(**{"name": "testColumn", "@id": "#test_col_id", "datatype": "double", "precision": 6})

1034

1035

1036class SchemaSerializationTest(unittest.TestCase):

1037 """Test serialization and deserialization of the schema data model."""

1038

1039 def test_serialization(self) -> None:

1040 """Test serialization of the schema data model."""

1041 # Read the original YAML content from the test_serialization.yaml file

1042 with open(TEST_SERIALIZATION) as file:

1043 original_yaml_content = file.read()

1044

1045 # Load the schema from the original YAML content

1046 schema_out = Schema.from_uri(TEST_SERIALIZATION)

1047 serialized_data = schema_out.model_dump(by_alias=True, exclude_none=True, exclude_defaults=True)

1048

1049 # Write the serialized data to a temporary YAML file

1050 with tempfile.NamedTemporaryFile(delete=False, suffix=".yaml", mode="w+") as temp_file:

1051 yaml.dump(serialized_data, temp_file, default_flow_style=False, sort_keys=False)

1052 temp_file.seek(0)

1053 # Read the deserialized YAML content from the temporary file

1054 deserialized_yaml_content = temp_file.read()

1055

1056 # Show the differences between the original and deserialized YAML

1057 diff = difflib.unified_diff(

1058 original_yaml_content.splitlines(keepends=True),

1059 deserialized_yaml_content.splitlines(keepends=True),

1060 fromfile="original.yaml",

1061 tofile="deserialized.yaml",

1062 )

1063 print("Differences:\n", "".join(diff))

1064

1065 # Assert that the original and deserialized YAML are the same

1066 self.assertEqual(

1067 yaml.safe_load(original_yaml_content),

1068 yaml.safe_load(deserialized_yaml_content),

1069 "The original and deserialized YAML contents should be the same",

1070 )

1071

1072

1073class ResourceTestCase(unittest.TestCase):

1074 """Test loading of column definitions from external schema resources."""

1075

1076 def setUp(self) -> None:

1077 """Set up test resources."""

1078 self.temp_dir = tempfile.mkdtemp()

1079

1080 # Write out source schema file

1081 source_schema_content = """

1082name: source_schema

1083description: Test resource schema

1084tables:

1085- name: source_table

1086 description: Source table

1087 columns:

1088 - name: test_column

1089 datatype: int

1090 description: "Test column"

1091"""

1092 self.source_schema_path = os.path.join(self.temp_dir, "source_schema.yaml")

1093 with open(self.source_schema_path, "w") as f:

1094 f.write(source_schema_content.strip())

1095

1096 # Write out referencing schema file

1097 ref_schema_content = """

1098name: ref_schema

1099description: Test referencing schema

1100resources:

1101 source_schema:

1102 uri: {resource_path}

1103tables:

1104- name: ref_table

1105 description: Referencing table

1106 columnRefs:

1107 source_schema:

1108 source_table:

1109 test_column: null # Explicit null = no overrides, use same name

1110 renamed_column:

1111 ref_name: test_column

1112 overrides:

1113 description: "Renamed test column"

1114 datatype: short

1115 tap:principal: 1

1116 tap:column_index: 2

1117"""

1118 self.ref_schema_path = os.path.join(self.temp_dir, "ref_schema.yaml")

1119 ref_content = ref_schema_content.format(resource_path=self.source_schema_path)

1120 with open(self.ref_schema_path, "w") as f:

1121 f.write(ref_content.strip())

1122

1123 def tearDown(self) -> None:

1124 """Clean up test resources."""

1125 shutil.rmtree(self.temp_dir)

1126

1127 def test_schema_resource(self) -> None:

1128 """Test loading a schema as a resource with column references."""

1129 # First test that the source schema loads correctly on its own

1130 source_schema = Schema.from_uri(self.source_schema_path, context={"id_generation": True})

1131 self.assertEqual(source_schema.name, "source_schema")

1132 self.assertEqual(len(source_schema.tables), 1)

1133 self.assertEqual(source_schema.tables[0].name, "source_table")

1134

1135 # Now test loading the ref schema

1136 ref_schema = Schema.from_uri(self.ref_schema_path, context={"id_generation": True})

1137 self.assertEqual(ref_schema.name, "ref_schema")

1138

1139 # Check that the resource was loaded

1140 self.assertIn("source_schema", ref_schema._resource_map)

1141

1142 # Check that the referencing table has the expected columns

1143 ref_table = ref_schema.tables[0]

1144 self.assertEqual(ref_table.name, "ref_table")

1145

1146 # Check the column_refs structure

1147 column_refs = ref_table.column_refs

1148 self.assertIsNotNone(column_refs)

1149 self.assertIsInstance(column_refs, dict)

1150

1151 # Check the schema resource reference

1152 self.assertIn("source_schema", column_refs)

1153 source_schema_refs = column_refs["source_schema"]

1154 self.assertIsInstance(source_schema_refs, dict)

1155

1156 # Check the table reference

1157 self.assertIn("source_table", source_schema_refs)

1158 source_table_refs = source_schema_refs["source_table"]

1159 self.assertIsInstance(source_table_refs, dict)

1160

1161 # Verify the column_refs structure details

1162 # Should have 2 column references: test_column and renamed_column

1163 self.assertEqual(len(source_table_refs), 2)

1164

1165 # Check test_column reference (null/no overrides)

1166 self.assertIn("test_column", source_table_refs)

1167 test_column_ref = source_table_refs["test_column"]

1168 self.assertIsNone(test_column_ref)

1169

1170 # Check renamed_column reference (with ref_name and overrides)

1171 self.assertIn("renamed_column", source_table_refs)

1172 renamed_column_ref = source_table_refs["renamed_column"]

1173 self.assertIsNotNone(renamed_column_ref)

1174 self.assertEqual(renamed_column_ref.ref_name, "test_column")

1175 self.assertIsNotNone(renamed_column_ref.overrides)

1176 self.assertEqual(renamed_column_ref.overrides.description, "Renamed test column")

1177 self.assertEqual(renamed_column_ref.overrides.tap_principal, 1)

1178 self.assertEqual(renamed_column_ref.overrides.tap_column_index, 2)

1179 self.assertEqual(renamed_column_ref.overrides.datatype.value, "short")

1180

1181 # Now check structure of dereferenced columns in the ref_table

1182 self.assertEqual(len(ref_table.columns), 2)

1183

1184 # Check dereferenced test_column (no overrides)

1185 test_col = next((col for col in ref_table.columns if col.name == "test_column"), None)

1186 self.assertIsNotNone(test_col)

1187 self.assertEqual(test_col.datatype, "int")

1188 self.assertEqual(test_col.description, "Test column")

1189

1190 # Check dereferenced renamed_column (includes overrides)

1191 renamed_col = next((col for col in ref_table.columns if col.name == "renamed_column"), None)

1192 self.assertIsNotNone(renamed_col)

1193 self.assertEqual(renamed_col.datatype, "short") # Inherited from source

1194 self.assertEqual(renamed_col.description, "Renamed test column")

1195 self.assertEqual(renamed_col.tap_principal, 1)

1196 self.assertEqual(renamed_col.tap_column_index, 2)

1197

1198 # Verify that the columns are present in the ID map

1199 try:

1200 ref_schema.find_object_by_id(test_col.id, Column)

1201 except KeyError:

1202 self.fail(f"Test column ID '{test_col.id}' not found in schema ID map.")

1203 try:

1204 ref_schema.find_object_by_id(renamed_col.id, Column)

1205 except KeyError:

1206 self.fail(f"Renamed column ID '{renamed_col.id}' not found in schema ID map.")

1207

1208 def test_schema_resource_missing_column_error(self) -> None:

1209 """Test that referencing a non-existent column raises an error."""

1210 error_ref_content = f"""

1211name: error_ref_schema

1212description: Test referencing schema with error

1213resources:

1214 source_schema:

1215 uri: {self.source_schema_path}

1216tables:

1217- name: error_table

1218 description: Table with bad reference

1219 columnRefs:

1220 source_schema:

1221 source_table:

1222 bad_column: null # This column doesn't exist in source

1223"""

1224

1225 error_ref_path = os.path.join(self.temp_dir, "error_ref_schema.yaml")

1226 with open(error_ref_path, "w") as f:

1227 f.write(error_ref_content.strip())

1228

1229 # This should raise a ValueError

1230 with self.assertRaises(ValueError) as cm:

1231 Schema.from_uri(error_ref_path, context={"id_generation": True})

1232

1233 self.assertIn("Column 'bad_column' not found", str(cm.exception))

1234

1235 def test_schema_resource_missing_ref_name_error(self) -> None:

1236 """Test that using ref_name for non-existent column raises an error."""

1237 # Create a ref schema with bad ref_name

1238 error_ref_content = f"""

1239name: error_ref_schema

1240description: Test referencing schema with bad ref_name

1241resources:

1242 source_schema:

1243 uri: {self.source_schema_path}

1244tables:

1245- name: error_table

1246 description: Table with bad ref_name

1247 columnRefs:

1248 source_schema:

1249 source_table:

1250 some_column:

1251 ref_name: nonexistent_column # This column doesn't exist

1252"""

1253

1254 error_ref_path = os.path.join(self.temp_dir, "error_ref_schema.yaml")

1255 with open(error_ref_path, "w") as f:

1256 f.write(error_ref_content.strip())

1257

1258 with self.assertRaises(ValueError) as cm:

1259 Schema.from_uri(error_ref_path, context={"id_generation": True})

1260 self.assertIn("Column 'nonexistent_column' not found", str(cm.exception))

1261

1262 def test_schema_resource_not_found_error(self) -> None:

1263 """Test that referencing a non-existent schema resource raises an

1264 error.

1265 """

1266 error_ref_content = f"""

1267name: error_ref_schema

1268description: Test referencing non-existent schema resource

1269resources:

1270 source_schema:

1271 uri: {self.source_schema_path}

1272tables:

1273- name: error_table

1274 description: Table with bad schema resource reference

1275 columnRefs:

1276 nonexistent_schema: # This schema resource doesn't exist

1277 some_table:

1278 some_column: null

1279"""

1280

1281 error_ref_path = os.path.join(self.temp_dir, "error_ref_schema.yaml")

1282 with open(error_ref_path, "w") as f:

1283 f.write(error_ref_content.strip())

1284

1285 with self.assertRaises(ValueError) as cm:

1286 Schema.from_uri(error_ref_path, context={"id_generation": True})

1287 self.assertIn("Schema resource 'nonexistent_schema' was not found in resources", str(cm.exception))

1288

1289 def test_schema_resource_table_not_found_error(self) -> None:

1290 """Test that referencing a non-existent table in schema resource raises

1291 an error.

1292 """

1293 error_ref_content = f"""

1294name: error_ref_schema

1295description: Test referencing non-existent table in schema resource

1296resources:

1297 source_schema:

1298 uri: {self.source_schema_path}

1299tables:

1300- name: error_table

1301 description: Table with bad table reference

1302 columnRefs:

1303 source_schema:

1304 nonexistent_table: # This table doesn't exist in source_schema

1305 some_column: null

1306"""

1307

1308 error_ref_path = os.path.join(self.temp_dir, "error_ref_schema.yaml")

1309 with open(error_ref_path, "w") as f:

1310 f.write(error_ref_content.strip())

1311

1312 with self.assertRaises(ValueError) as cm:

1313 Schema.from_uri(error_ref_path, context={"id_generation": True})

1314 self.assertIn("Table 'nonexistent_table' not found in resource 'source_schema'", str(cm.exception))

1315

1316 def test_schema_resource_bad_uri_error(self) -> None:

1317 """Test that a bad URI in resource loading raises an error."""

1318 error_ref_content = """

1319name: error_ref_schema

1320description: Test schema with bad resource URI

1321resources:

1322 bad_resource:

1323 uri: /nonexistent/path/to/schema.yaml

1324tables:

1325- name: error_table

1326 description: Table referencing bad resource

1327 columnRefs:

1328 bad_resource:

1329 some_table:

1330 some_column: null

1331"""

1332

1333 error_ref_path = os.path.join(self.temp_dir, "error_ref_schema.yaml")

1334 with open(error_ref_path, "w") as f:

1335 f.write(error_ref_content.strip())

1336

1337 with self.assertRaises(ValueError) as cm:

1338 Schema.from_uri(error_ref_path, context={"id_generation": True})

1339 self.assertIn(

1340 "Failed to load resource 'bad_resource' from URI '/nonexistent/path/to/schema.yaml'",

1341 str(cm.exception),

1342 )

1343

1344 def test_ref_schema_with_indexes(self) -> None:

1345 """Test that indexes are properly handled when loading schema

1346 resources.

1347 """

1348 # Write out referencing schema file

1349 ref_schema_content_with_indexes = """

1350name: ref_schema

1351description: Test referencing schema

1352resources:

1353 source_schema:

1354 uri: {resource_path}

1355tables:

1356- name: ref_table

1357 description: Referencing table

1358 columnRefs:

1359 source_schema:

1360 source_table:

1361 test_column: null

1362 renamed_column:

1363 ref_name: test_column

1364 overrides:

1365 description: "Renamed test column"

1366 tap:principal: 1

1367 tap:column_index: 2

1368 indexes:

1369 - name: idx_test_column

1370 columns:

1371 - "#ref_table.test_column"

1372 - name: idx_renamed_column

1373 columns:

1374 - "#ref_table.renamed_column"

1375"""

1376

1377 source_schema_with_indexes_path = os.path.join(self.temp_dir, "source_schema_with_indexes.yaml")

1378 ref_content = ref_schema_content_with_indexes.format(resource_path=self.source_schema_path)

1379 with open(source_schema_with_indexes_path, "w") as f:

1380 f.write(ref_content.strip())

1381

1382 ref_schema = Schema.from_uri(source_schema_with_indexes_path, context={"id_generation": True})

1383

1384 # Check index content; columns are not automatically resolved to

1385 # objects by the validation.

1386 indexes = ref_schema.tables[0].indexes

1387 self.assertEqual(len(indexes), 2)

1388 self.assertEqual(indexes[0].name, "idx_test_column")

1389 self.assertEqual(indexes[0].columns, ["#ref_table.test_column"])

1390 self.assertEqual(indexes[1].name, "idx_renamed_column")

1391 self.assertEqual(indexes[1].columns, ["#ref_table.renamed_column"])

1392

1393 def test_ref_schema_with_foreign_key(self) -> None:

1394 """Test that indexes are properly handled when loading schema

1395 resources.

1396 """

1397 # Write out referencing schema file

1398 ref_schema_content_with_foreign_key = """

1399name: ref_schema

1400description: Test referencing schema

1401resources:

1402 source_schema:

1403 uri: {resource_path}

1404tables:

1405- name: src_table

1406 description: Source table for foreign key

1407 primaryKey: "#src_table.test_column"

1408 columnRefs:

1409 source_schema:

1410 source_table:

1411 test_column: null

1412 renamed_column:

1413 ref_name: test_column

1414 overrides:

1415 description: "Renamed test column"

1416 tap:principal: 1

1417 tap:column_index: 2

1418- name: target_table

1419 description: Target table for foreign key

1420 columns:

1421 - name: fk_column

1422 datatype: int

1423 description: "Foreign key column"

1424 constraints:

1425 - name: fk_src_table

1426 '@type': ForeignKey

1427 columns:

1428 - "#target_table.fk_column"

1429 referencedColumns:

1430 - "#src_table.test_column"

1431"""

1432

1433 source_schema_with_foreign_key_path = os.path.join(

1434 self.temp_dir, "source_schema_with_foreign_key.yaml"

1435 )

1436 ref_content = ref_schema_content_with_foreign_key.format(resource_path=self.source_schema_path)

1437 with open(source_schema_with_foreign_key_path, "w") as f:

1438 f.write(ref_content.strip())

1439

1440 ref_schema = Schema.from_uri(source_schema_with_foreign_key_path, context={"id_generation": True})

1441

1442 # Check foreign key constraint content

1443 fk_constraint = ref_schema.tables[1].constraints[0]

1444 self.assertIsInstance(fk_constraint, ForeignKeyConstraint)

1445 self.assertEqual(fk_constraint.name, "fk_src_table")

1446 self.assertEqual(fk_constraint.columns, ["#target_table.fk_column"])

1447 self.assertEqual(fk_constraint.referenced_columns, ["#src_table.test_column"])

1448

1449 def test_ref_schema_serialization(self) -> None:

1450 """Test serialization of a reference schema."""

1451 # Load the referencing schema and then serialize it back to YAML

1452 ref_schema = Schema.from_uri(self.ref_schema_path, context={"id_generation": True})

1453 yaml_data = ref_schema.model_dump(by_alias=True, exclude_none=True, exclude_defaults=True)

1454 serialized_schema_path = os.path.join(self.temp_dir, "serialized_ref_schema.yaml")

1455 with open(serialized_schema_path, "w") as f:

1456 yaml.dump(yaml_data, f, default_flow_style=False, sort_keys=False)

1457

1458 # Read back the serialized YAML data

1459 with open(serialized_schema_path) as f:

1460 serialized_yaml_data = yaml.safe_load(f)

1461

1462 # Ensure that columns were not serialized directly in the table

1463 self.assertEqual(len(serialized_yaml_data["tables"][0]["columns"]), 0)

1464

1465 # Deserialize the schema and check that the expected columns are

1466 # present

1467 deserialized_schema = Schema.from_uri(serialized_schema_path, context={"id_generation": True})

1468 self.assertEqual(len(deserialized_schema.tables[0].columns), 2)

1469

1470 # Check that the columnRefs structure is still present

1471 try:

1472 ref_columns = deserialized_schema.tables[0].column_refs["source_schema"]["source_table"]

1473 except Exception:

1474 self.fail("The column refs are missing after deserialization.")

1475 self.assertEqual(len(ref_columns), 2)

1476

1477 def test_ref_schema_with_dereference_columns(self) -> None:

1478 """Test loading a reference schema with dereferencing of columns so

1479 that column_refs is set to empty after loading.

1480 """

1481 ref_schema = Schema.from_uri(

1482 self.ref_schema_path, context={"id_generation": True, "dereference_resources": True}

1483 )

1484

1485 # Check that the columns were dereferenced into the table

1486 ref_table = ref_schema.tables[0]

1487 self.assertEqual(len(ref_table.columns), 2)

1488 col_names = {col.name for col in ref_table.columns}

1489 self.assertIn("test_column", col_names)

1490 self.assertIn("renamed_column", col_names)

1491

1492 # Check that column_refs is empty after dereferencing

1493 self.assertEqual(len(ref_table.column_refs), 0)

1494

1495 def test_tap_column_index_with_overrides(self) -> None:

1496 """Test that TAP column index is correctly assigned when an override

1497 of that field is present in the column ref.

1498 """

1499 # Write out source schema file

1500 source_schema_content = """

1501name: source_schema

1502tables:

1503- name: source_table

1504 columns:

1505 - name: col1

1506 datatype: int

1507 - name: col2

1508 datatype: int

1509 - name: col3

1510 datatype: int

1511"""

1512 source_schema_path = os.path.join(self.temp_dir, "source_schema.yaml")

1513 with open(source_schema_path, "w") as f:

1514 f.write(source_schema_content.strip())

1515

1516 # Write out referencing schema file

1517 ref_schema_content = """

1518name: ref_schema

1519resources:

1520 source_schema:

1521 uri: {resource_path}

1522tables:

1523- name: ref_table

1524 columnRefs:

1525 source_schema:

1526 source_table:

1527 col1:

1528 col2:

1529 overrides:

1530 tap:column_index: 15

1531 col3:

1532"""

1533 ref_schema_path = os.path.join(self.temp_dir, "ref_schema.yaml")

1534 ref_content = ref_schema_content.format(resource_path=source_schema_path)

1535 with open(ref_schema_path, "w") as f:

1536 f.write(ref_content.strip())

1537

1538 ref_schema = Schema.from_uri(

1539 ref_schema_path,

1540 context={"id_generation": True, "column_ref_index_increment": 10},

1541 )

1542

1543 for column in ref_schema.tables[0].columns:

1544 if column.name == "col1":

1545 self.assertEqual(column.tap_column_index, 10)

1546 elif column.name == "col2":

1547 self.assertEqual(column.tap_column_index, 15)

1548 elif column.name == "col3":

1549 self.assertEqual(column.tap_column_index, 20)

1550 else:

1551 self.fail(f"Unexpected column name: {column.name}")

1552

1553

1554if __name__ == "__main__": 1554 ↛ 1555line 1554 didn't jump to line 1555 because the condition on line 1554 was never true

1555 unittest.main()

Coverage for tests / test_datamodel.py: 11%

647 statements