Coverage for tests/test_cli.py: 94%
267 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-27 01:40 -0700
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 logging
23import os
24import shutil
25import tempfile
26import unittest
27from typing import Any
29import yaml
30from sqlalchemy import create_engine, text
32import felis.tap_schema as tap_schema
33from felis.datamodel import Schema
34from felis.metadata import MetaDataBuilder
35from felis.tests.run_cli import run_cli
37TEST_DIR = os.path.abspath(os.path.dirname(__file__))
38TEST_YAML = os.path.join(TEST_DIR, "data", "test.yml")
39TEST_SALES_YAML = os.path.join(TEST_DIR, "data", "sales.yaml")
42class CliTestCase(unittest.TestCase):
43 """Tests for CLI commands."""
45 def setUp(self) -> None:
46 """Set up a temporary directory for tests."""
47 self.tmpdir = tempfile.mkdtemp(dir=TEST_DIR)
48 self.sqlite_url = f"sqlite:///{self.tmpdir}/db.sqlite3"
49 print(f"Using temporary directory: {self.tmpdir}")
51 def tearDown(self) -> None:
52 """Clean up temporary directory."""
53 shutil.rmtree(self.tmpdir, ignore_errors=True)
55 def test_invalid_command(self) -> None:
56 """Test for invalid command."""
57 run_cli(["invalid"], expect_error=True)
59 def test_help(self) -> None:
60 """Test for help command."""
61 run_cli(["--help"], print_output=True)
63 def test_create(self) -> None:
64 """Test for create command."""
65 run_cli(["create", f"--engine-url={self.sqlite_url}", TEST_YAML])
67 def test_create_with_echo(self) -> None:
68 """Test for create command."""
69 run_cli(["create", "--echo", f"--engine-url={self.sqlite_url}", TEST_YAML])
71 def test_create_with_dry_run(self) -> None:
72 """Test for ``create --dry-run`` command."""
73 run_cli(["create", "--schema-name=main", f"--engine-url={self.sqlite_url}", "--dry-run", TEST_YAML])
75 def test_create_with_ignore_constraints(self) -> None:
76 """Test ``--ignore-constraints`` flag of ``create`` command."""
77 run_cli(
78 [
79 "create",
80 "--schema-name=main",
81 "--ignore-constraints",
82 f"--engine-url={self.sqlite_url}",
83 "--dry-run",
84 TEST_YAML,
85 ]
86 )
88 def test_browser_with_config(self) -> None:
89 """Test browser command with a site configuration file."""
90 output_dir = os.path.join(self.tmpdir, "site")
91 for name in ("theme.css", "local.css"):
92 with open(os.path.join(self.tmpdir, name), "w", encoding="utf-8") as f:
93 f.write(":root { --color-link: red; }\n")
94 config_path = os.path.join(self.tmpdir, "site.yaml")
95 with open(config_path, "w", encoding="utf-8") as f:
96 f.write("title: SDM Schemas\nstylesheets:\n - theme.css\n - local.css\n")
97 run_cli(["browser", "--output-dir", output_dir, "--config", config_path, TEST_YAML])
98 self.assertTrue(os.path.exists(os.path.join(output_dir, "assets", "custom-1-theme.css")))
99 self.assertTrue(os.path.exists(os.path.join(output_dir, "assets", "custom-2-local.css")))
100 with open(os.path.join(output_dir, "index.html"), encoding="utf-8") as f:
101 self.assertIn("<title>SDM Schemas</title>", f.read())
103 def test_browser_with_invalid_config(self) -> None:
104 """Test browser command with an invalid site configuration file."""
105 config_path = os.path.join(self.tmpdir, "site.yaml")
106 with open(config_path, "w", encoding="utf-8") as f:
107 f.write("unknown: setting\n")
108 run_cli(
109 [
110 "browser",
111 "--output-dir",
112 os.path.join(self.tmpdir, "site"),
113 "--config",
114 config_path,
115 TEST_YAML,
116 ],
117 expect_error=True,
118 )
120 def test_browser_with_malformed_config(self) -> None:
121 """Test browser command with a site configuration file that isn't
122 valid YAML: it fails with an error message, not a traceback.
123 """
124 config_path = os.path.join(self.tmpdir, "site.yaml")
125 with open(config_path, "w", encoding="utf-8") as f:
126 f.write("title: [unclosed\n")
127 run_cli(
128 [
129 "browser",
130 "--output-dir",
131 os.path.join(self.tmpdir, "site"),
132 "--config",
133 config_path,
134 TEST_YAML,
135 ],
136 expect_error=True,
137 )
139 def test_browser_resource(self) -> None:
140 """Test browser command with a schema given as a resource URI."""
141 output_dir = os.path.join(self.tmpdir, "site")
142 run_cli(
143 [
144 "browser",
145 "--output-dir",
146 output_dir,
147 "resource://felis/config/tap_schema/tap_schema_std.yaml",
148 ],
149 id_generation=True,
150 )
151 self.assertTrue(os.path.isfile(os.path.join(output_dir, "schemas", "tap-schema", "index.html")))
153 def test_browser_missing_file(self) -> None:
154 """Test browser command with a schema file that doesn't exist."""
155 run_cli(
156 [
157 "browser",
158 "--output-dir",
159 os.path.join(self.tmpdir, "site"),
160 os.path.join(self.tmpdir, "missing.yaml"),
161 ],
162 expect_error=True,
163 )
165 def test_browser_unresolved_column(self) -> None:
166 """Test browser command with an index on a column that isn't in the
167 schema: it fails with an error message.
168 """
169 schema_path = os.path.join(self.tmpdir, "refs.yaml")
170 with open(schema_path, "w", encoding="utf-8") as f:
171 f.write(
172 "name: refs\n"
173 "'@id': '#refs'\n"
174 "tables:\n"
175 " - name: data\n"
176 " '@id': '#data'\n"
177 " columns:\n"
178 " - {name: id, '@id': '#data.id', datatype: int, description: Key}\n"
179 " indexes:\n"
180 " - {name: idx_missing, '@id': '#idx_missing', columns: ['#data.missing']}\n"
181 )
182 run_cli(
183 ["browser", "--output-dir", os.path.join(self.tmpdir, "site"), schema_path], expect_error=True
184 )
186 def test_browser_non_yaml_file(self) -> None:
187 """Test browser command with an input file that isn't a schema."""
188 bad_file = os.path.join(self.tmpdir, "not_yaml.txt")
189 with open(bad_file, "w", encoding="utf-8") as f:
190 f.write("hello")
191 run_cli(
192 ["browser", "--output-dir", os.path.join(self.tmpdir, "site"), bad_file],
193 expect_error=True,
194 )
196 def test_validate(self) -> None:
197 """Test validate command."""
198 run_cli(["validate", TEST_YAML])
200 def test_validate_with_log_file(self) -> None:
201 """Test validate command with log file."""
202 log_file = os.path.join(self.tmpdir, "validate.log")
203 run_cli([f"--log-file={log_file}", "validate", TEST_YAML], log_level=logging.DEBUG, print_cmd=True)
204 if not os.path.exists(log_file): 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true
205 self.fail("Log file was not created")
206 if os.path.getsize(log_file) == 0: 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 self.fail("Log file is empty")
209 def test_validate_with_id_generation(self) -> None:
210 """Test that loading a schema with IDs works if ID generation is
211 enabled. This is the default behavior.
212 """
213 test_yaml = os.path.join(TEST_DIR, "data", "test_id_generation.yaml")
214 run_cli(["--id-generation", "validate", test_yaml])
216 def test_validate_with_id_generation_error(self) -> None:
217 """Test that loading a schema without IDs fails if ID generation is not
218 enabled.
219 """
220 test_yaml = os.path.join(TEST_DIR, "data", "test_id_generation.yaml")
221 run_cli(["--no-id-generation", "validate", test_yaml], expect_error=True)
223 def test_validate_with_extra_checks(self) -> None:
224 """Test schema validation flags."""
225 run_cli(
226 [
227 "validate",
228 "--check-description",
229 "--check-tap-principal",
230 "--check-tap-table-indexes",
231 TEST_YAML,
232 ]
233 )
235 def test_create_with_initialize_and_drop_error(self) -> None:
236 """Test that initialize and drop can't be used together."""
237 run_cli(["create", "--initialize", "--drop", TEST_YAML], expect_error=True)
239 def test_load_tap_schema(self) -> None:
240 """Test load-tap-schema command."""
241 # Create the TAP_SCHEMA database.
242 tap_schema_path = tap_schema.TableManager.get_tap_schema_std_path()
243 run_cli(["--id-generation", "create", f"--engine-url={self.sqlite_url}", tap_schema_path])
245 # Load the TAP_SCHEMA data.
246 run_cli(["load-tap-schema", f"--engine-url={self.sqlite_url}", TEST_YAML])
248 def test_load_tap_schema_with_dry_run_and_output_file(self) -> None:
249 """Test load-tap-schema command with dry run and output file."""
250 output_sql = os.path.join(self.tmpdir, "tap_schema.sql")
251 run_cli(
252 [
253 "load-tap-schema",
254 "--engine-url=mysql://",
255 "--dry-run",
256 "--tap-schema-index=1",
257 "--tap-tables-postfix=11",
258 "--force-unbounded-arraysize",
259 f"--output-file={output_sql}",
260 TEST_YAML,
261 ]
262 )
263 if not os.path.exists(output_sql): 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true
264 self.fail("Output SQL file was not created")
265 if os.path.getsize(output_sql) == 0: 265 ↛ 266line 265 didn't jump to line 266 because the condition on line 265 was never true
266 self.fail("Output SQL file is empty")
268 def test_init_tap_schema(self) -> None:
269 """Test init-tap-schema command."""
270 run_cli(["init-tap-schema", f"--engine-url={self.sqlite_url}"])
272 def test_init_tap_schema_mock(self) -> None:
273 """Test init-tap-schema command with a mock URL, which should throw
274 an error, as this is not supported.
275 """
276 run_cli(["init-tap-schema", "sqlite://"], expect_error=True)
278 def test_init_tap_schema_with_extensions(self) -> None:
279 """Test init-tap-schema command with default extensions."""
280 run_cli(
281 [
282 "init-tap-schema",
283 f"--engine-url={self.sqlite_url}",
284 "--extensions",
285 "resource://felis/config/tap_schema/tap_schema_extensions.yaml",
286 ]
287 )
289 def test_init_tap_schema_with_custom_extensions(self) -> None:
290 """Test init-tap-schema command with custom extensions file."""
291 extensions_file = os.path.join(self.tmpdir, "custom_extensions.yaml")
292 extensions_content = """
293 name: TAP_SCHEMA
294 tables:
295 - name: schemas
296 columns:
297 - name: field1
298 datatype: char
299 length: 64
300 description: A custom field
301 """
302 with open(extensions_file, "w") as f:
303 f.write(extensions_content)
305 run_cli(["init-tap-schema", f"--engine-url={self.sqlite_url}", "--extensions", extensions_file])
307 def test_diff(self) -> None:
308 """Test for ``diff`` command."""
309 test_diff1 = os.path.join(TEST_DIR, "data", "test_diff1.yaml")
310 test_diff2 = os.path.join(TEST_DIR, "data", "test_diff2.yaml")
312 run_cli(["diff", test_diff1, test_diff2])
314 def test_diff_database(self) -> None:
315 """Test for ``diff`` command with database."""
316 test_diff1 = os.path.join(TEST_DIR, "data", "test_diff1.yaml")
317 test_diff2 = os.path.join(TEST_DIR, "data", "test_diff2.yaml")
319 engine = create_engine(self.sqlite_url)
320 metadata_db = MetaDataBuilder(Schema.from_uri(test_diff1), apply_schema_to_metadata=False).build()
321 metadata_db.create_all(engine)
322 engine.dispose()
324 run_cli(["diff", f"--engine-url={self.sqlite_url}", test_diff2])
326 def test_diff_alembic(self) -> None:
327 """Test for ``diff`` command with ``--alembic`` comparator option."""
328 test_diff1 = os.path.join(TEST_DIR, "data", "test_diff1.yaml")
329 test_diff2 = os.path.join(TEST_DIR, "data", "test_diff2.yaml")
330 run_cli(["diff", "--comparator", "alembic", test_diff1, test_diff2], print_output=True)
332 def test_diff_error(self) -> None:
333 """Test for ``diff`` command with bad arguments."""
334 test_diff1 = os.path.join(TEST_DIR, "data", "test_diff1.yaml")
335 run_cli(["diff", test_diff1], expect_error=True)
337 def test_diff_error_on_change(self) -> None:
338 """Test for ``diff`` command with ``--error-on-change`` flag."""
339 test_diff1 = os.path.join(TEST_DIR, "data", "test_diff1.yaml")
340 test_diff2 = os.path.join(TEST_DIR, "data", "test_diff2.yaml")
341 run_cli(["diff", "--error-on-change", test_diff1, test_diff2], expect_error=True, print_output=True)
343 def test_dump_yaml(self) -> None:
344 """Test for ``dump`` command with YAML output."""
345 with tempfile.NamedTemporaryFile(delete=True, suffix=".yaml") as temp_file:
346 run_cli(["dump", TEST_YAML, temp_file.name], print_output=True)
348 @classmethod
349 def _check_strip_ids(cls, obj: Any) -> None:
350 """
351 Recursively check that a dict/list structure has no attributes with key
352 '@id'. Raises a ValueError if any '@id' key is found. This is used to
353 check the output of the `--strip-ids` option in the `dump` command.
354 """
355 if isinstance(obj, dict):
356 for k, v in obj.items():
357 if k == "@id": 357 ↛ 358line 357 didn't jump to line 358 because the condition on line 357 was never true
358 raise ValueError("Found forbidden key '@id'")
359 cls._check_strip_ids(v)
360 elif isinstance(obj, list):
361 for item in obj:
362 cls._check_strip_ids(item)
364 def test_dump_yaml_with_strip_ids(self) -> None:
365 """Test for ``dump`` command with YAML output and stripped IDs."""
366 with tempfile.NamedTemporaryFile(delete=True, suffix=".yaml") as temp_file:
367 run_cli(["dump", "--strip-ids", TEST_YAML, temp_file.name], print_output=True)
368 dumped_data = temp_file.read().decode("utf-8")
369 try:
370 # Load the dumped YAML data to check for '@id' keys.
371 data = yaml.safe_load(dumped_data)
372 self._check_strip_ids(data)
373 except ValueError:
374 self.fail("Dumped YAML contains forbidden key '@id'")
376 def test_dump_json(self) -> None:
377 """Test for ``dump`` command with JSON output."""
378 with tempfile.NamedTemporaryFile(delete=True, suffix=".json") as temp_file:
379 run_cli(["dump", TEST_YAML, temp_file.name], print_output=True)
381 def test_dump_with_dereference_resources_and_sort_columns(self) -> None:
382 """Test dump with both --dereference-resources and --sort-columns."""
383 # Define a source schema with columns in non-alphabetical order
384 source_schema_content = """
385name: base_schema
386tables:
387- name: base_table
388 columns:
389 - name: zebra_col
390 datatype: string
391 length: 32
392 - name: alpha_col
393 datatype: int
394 - name: middle_col
395 datatype: float
396"""
397 source_path = os.path.join(self.tmpdir, "base_schema.yaml")
398 with open(source_path, "w") as f:
399 f.write(source_schema_content.strip())
401 # Define a referencing schema that pulls columns via columnRefs
402 ref_schema_content = f"""
403name: derived_schema
404resources:
405 base_schema:
406 uri: {source_path}
407tables:
408- name: derived_table
409 columnRefs:
410 base_schema:
411 base_table:
412 zebra_col:
413 alpha_col:
414 middle_col:
415"""
416 ref_path = os.path.join(self.tmpdir, "derived_schema.yaml")
417 with open(ref_path, "w") as f:
418 f.write(ref_schema_content.strip())
420 with tempfile.NamedTemporaryFile(delete=True, suffix=".yaml", dir=self.tmpdir) as temp_file:
421 run_cli(
422 [
423 "dump",
424 "--dereference-resources",
425 "--sort-columns",
426 ref_path,
427 temp_file.name,
428 ],
429 print_output=True,
430 )
431 dumped_data = temp_file.read().decode("utf-8")
432 data = yaml.safe_load(dumped_data)
434 # Verify resources are dereferenced (no columnRefs remain)
435 for table in data.get("tables", []):
436 self.assertNotIn("columnRefs", table)
437 # Verify columns are present and sorted alphabetically
438 columns = table.get("columns", [])
439 self.assertGreater(len(columns), 0)
440 names = [col["name"] for col in columns]
441 self.assertEqual(names, sorted(names))
443 def test_dump_json_with_strip_ids(self) -> None:
444 """Test for ``dump`` command with JSON output."""
445 with tempfile.NamedTemporaryFile(delete=True, suffix=".json") as temp_file:
446 run_cli(["dump", "--strip-ids", TEST_YAML, temp_file.name], print_output=True)
447 dumped_data = temp_file.read().decode("utf-8")
448 try:
449 # Load the dumped YAML data to check for '@id' keys.
450 data = yaml.safe_load(dumped_data)
451 self._check_strip_ids(data)
452 except ValueError:
453 self.fail("Dumped YAML contains forbidden key '@id'")
455 @classmethod
456 def _check_columns_sorted(cls, data: dict[str, Any]) -> None:
457 """Check that columns in each table are sorted alphabetically by
458 name.
459 """
460 for table in data.get("tables", []):
461 columns = table.get("columns", [])
462 names = [col["name"] for col in columns]
463 assert names == sorted(names), f"Columns not sorted in table {table.get('name')}: {names}"
465 def test_dump_yaml_with_sort_columns(self) -> None:
466 """Test for ``dump`` command with YAML output and sorted columns."""
467 with tempfile.NamedTemporaryFile(delete=True, suffix=".yaml") as temp_file:
468 run_cli(["dump", "--sort-columns", TEST_YAML, temp_file.name], print_output=True)
469 dumped_data = temp_file.read().decode("utf-8")
470 data = yaml.safe_load(dumped_data)
471 self._check_columns_sorted(data)
473 def test_dump_json_with_sort_columns(self) -> None:
474 """Test for ``dump`` command with JSON output and sorted columns."""
475 with tempfile.NamedTemporaryFile(delete=True, suffix=".json") as temp_file:
476 run_cli(["dump", "--sort-columns", TEST_YAML, temp_file.name], print_output=True)
477 dumped_data = temp_file.read().decode("utf-8")
478 data = yaml.safe_load(dumped_data)
479 self._check_columns_sorted(data)
481 def test_dump_with_invalid_file_extension_error(self) -> None:
482 """Test for ``dump`` command with JSON output."""
483 run_cli(["dump", TEST_YAML, "out.bad"], expect_error=True)
485 def test_create_and_drop_indexes(self) -> None:
486 """Test creating and dropping indexes using CLI commands with
487 SQLite; no checking for the existence of the indexes is done on the
488 database because other test cases cover that functionality
489 sufficiently.
490 """
491 # Create database without indexes
492 run_cli(["create", "--skip-indexes", f"--engine-url={self.sqlite_url}", TEST_SALES_YAML])
494 # Create the indexes using CLI
495 run_cli(
496 ["create-indexes", f"--engine-url={self.sqlite_url}", TEST_SALES_YAML, "--schema-name", "main"]
497 )
499 # Create the indexes again; should not cause an error
500 run_cli(
501 ["create-indexes", f"--engine-url={self.sqlite_url}", TEST_SALES_YAML, "--schema-name", "main"]
502 )
504 # Drop the indexes using CLI
505 run_cli(["drop-indexes", f"--engine-url={self.sqlite_url}", TEST_SALES_YAML, "--schema-name", "main"])
507 def test_generate_and_load_sql(self) -> None:
508 """Test generating SQL and then executing it on a SQLite database."""
509 generated_sql = os.path.join(self.tmpdir, "generated.sql")
511 try:
512 # Generate SQL DDL from schema using mock connection
513 run_cli(
514 [
515 "create",
516 "--engine-url=sqlite://",
517 f"--output-file={generated_sql}",
518 f"{TEST_YAML}",
519 ]
520 )
522 # Verify the SQL file was generated
523 self.assertTrue(os.path.exists(generated_sql), "Generated SQL file should exist")
525 # Read the generated SQL
526 with open(generated_sql) as f:
527 sql = f.read()
529 # Verify SQL content is not empty
530 self.assertGreater(len(sql.strip()), 0, "Generated SQL should not be empty")
532 # Execute the SQL against a real database
533 engine = create_engine(self.sqlite_url)
534 with engine.connect() as connection:
535 with connection.begin():
536 # Split SQL into individual statements for execution since
537 # SQLite can only execute one statement at a time
538 statements = [stmt.strip() for stmt in sql.split(";") if stmt.strip()]
539 for statement in statements:
540 if statement: # Skip empty statements 540 ↛ 539line 540 didn't jump to line 539 because the condition on line 540 was always true
541 connection.execute(text(statement))
543 # Verify that all expected tables were actually created
544 with engine.connect() as connection:
545 # Load the schema to get expected table names
546 schema = Schema.from_uri(TEST_YAML, context={"id_generation": True})
547 expected_table_names = {table.name for table in schema.tables}
549 # Get all tables that were created in the database
550 result = connection.execute(text("SELECT name FROM sqlite_master WHERE type='table'"))
551 created_table_names = {row[0] for row in result.fetchall()}
553 # Verify all expected tables were created
554 self.assertTrue(
555 expected_table_names.issubset(created_table_names),
556 f"Missing tables: {expected_table_names - created_table_names}. "
557 f"Expected: {sorted(expected_table_names)}, "
558 f"Created: {sorted(created_table_names)}",
559 )
561 engine.dispose()
563 except Exception as e:
564 self.fail(f"Test failed with exception: {e}")
567class ColumnRefsTestCase(unittest.TestCase):
568 """Test handling of column references in CLI."""
570 def setUp(self) -> None:
571 """Set up a temporary directory for tests."""
572 self.temp_dir = tempfile.mkdtemp(dir=TEST_DIR)
573 self.sqlite_url = f"sqlite:///{self.temp_dir}/db.sqlite3"
575 # Write out source schema file
576 source_schema_content = """
577name: source_schema
578tables:
579- name: source_table
580 columns:
581 - name: ref_col1
582 datatype: int
583 - name: ref_col2
584 datatype: string
585 length: 64
586 - name: ref_col3
587 datatype: float
588"""
589 source_schema_path = os.path.join(self.temp_dir, "source_schema.yaml")
590 with open(source_schema_path, "w") as f:
591 f.write(source_schema_content.strip())
593 # Write out referencing schema file
594 ref_schema_content = """
595name: ref_schema
596resources:
597 source_schema:
598 uri: {resource_path}
599tables:
600- name: ref_table
601 columnRefs:
602 source_schema:
603 source_table:
604 ref_col1:
605 ref_col2:
606 overrides:
607 tap:column_index: 15
608 col3:
609 ref_name: ref_col3
610"""
611 self.ref_schema_path = os.path.join(self.temp_dir, "ref_schema.yaml")
612 ref_content = ref_schema_content.format(resource_path=source_schema_path)
613 with open(self.ref_schema_path, "w") as f:
614 f.write(ref_content.strip())
616 def tearDown(self) -> None:
617 """Clean up temporary directory."""
618 shutil.rmtree(self.temp_dir, ignore_errors=True)
620 def test_validate_with_column_ref_index_increment(self) -> None:
621 """Test that passing a valid value for column reference index increment
622 works.
623 """
624 run_cli(
625 [
626 "--column-ref-index-increment=1",
627 "validate",
628 self.ref_schema_path,
629 ]
630 )
632 def test_validate_with_column_ref_index_increment_error(self) -> None:
633 """Test that passing an invalid value for column reference index raises
634 an error.
635 """
636 run_cli(
637 [
638 "--column-ref-index-increment=-1",
639 "validate",
640 self.ref_schema_path,
641 ],
642 expect_error=True,
643 )
645 def test_load_tap_schema_with_column_refs(self) -> None:
646 """Test load-tap-schema command with column reference index
647 increment.
648 """
649 # Create the TAP_SCHEMA database
650 run_cli(["init-tap-schema", f"--engine-url={self.sqlite_url}"])
652 # Load the TAP_SCHEMA data that includes column references
653 run_cli(
654 [
655 "load-tap-schema",
656 f"--engine-url={self.sqlite_url}",
657 self.ref_schema_path,
658 ]
659 )
661 def test_load_tap_schema_with_column_ref_index_increment(self) -> None:
662 """Test load-tap-schema command with column reference index
663 increment.
664 """
665 # Create the TAP_SCHEMA database
666 run_cli(["init-tap-schema", f"--engine-url={self.sqlite_url}"])
668 # Load the TAP_SCHEMA data that includes column reference index
669 # increment
670 run_cli(
671 [
672 "--column-ref-index-increment=1",
673 "load-tap-schema",
674 f"--engine-url={self.sqlite_url}",
675 self.ref_schema_path,
676 ]
677 )
680if __name__ == "__main__": 680 ↛ 681line 680 didn't jump to line 681 because the condition on line 680 was never true
681 unittest.main()