106 lines
2.5 KiB
Python
106 lines
2.5 KiB
Python
import json
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app
|
|
from tht.config import load_config
|
|
|
|
|
|
def _write_old_workspace(tmp_path):
|
|
path = tmp_path / "old.yaml"
|
|
path.write_text(
|
|
"""
|
|
database:
|
|
host: ignored-for-rest
|
|
database: analytics
|
|
schema: mart
|
|
user: legacy-user
|
|
password: legacy-password
|
|
transport: rest
|
|
rest:
|
|
base_url: https://dwh.example.test/
|
|
api_key: dwh-reader
|
|
vector_db:
|
|
host: vector-db
|
|
database: postgres
|
|
schema: vectors
|
|
user: vector-user
|
|
password: vector-password
|
|
vector_rest:
|
|
base_url: https://vectors.example.test/
|
|
api_key: vector-reader
|
|
vector_write_rest:
|
|
base_url: https://vectors.example.test/
|
|
api_key: vector-writer
|
|
paths:
|
|
artifacts: build/artifacts
|
|
indexes: build/indexes
|
|
sessions: build/sessions
|
|
"""
|
|
)
|
|
return path
|
|
|
|
|
|
def _write_new_workspace(tmp_path):
|
|
path = tmp_path / "new.yaml"
|
|
path.write_text(
|
|
"""
|
|
dwh:
|
|
type: thoth_rest
|
|
database:
|
|
database: analytics
|
|
schema: mart
|
|
endpoint:
|
|
base_url: https://dwh.example.test/
|
|
api_key: dwh-reader
|
|
vectors:
|
|
type: thoth_vector_http
|
|
reader:
|
|
base_url: https://vectors.example.test/
|
|
api_key: vector-reader
|
|
writer:
|
|
base_url: https://vectors.example.test/
|
|
api_key: vector-writer
|
|
direct:
|
|
host: vector-db
|
|
database: postgres
|
|
schema: vectors
|
|
user: vector-user
|
|
password: vector-password
|
|
roots:
|
|
artifacts: build/artifacts
|
|
indexes: build/indexes
|
|
sessions: build/sessions
|
|
"""
|
|
)
|
|
return path
|
|
|
|
|
|
def test_legacy_rest_workspace_equals_new_resource_schema(tmp_path, capsys):
|
|
with pytest.warns(FutureWarning, match="DEPRECATION") as warnings:
|
|
old = load_config(_write_old_workspace(tmp_path))
|
|
captured = capsys.readouterr()
|
|
new = load_config(_write_new_workspace(tmp_path))
|
|
|
|
assert old.dwh.model_dump() == new.dwh.model_dump()
|
|
assert old.vectors.model_dump() == new.vectors.model_dump()
|
|
assert old.roots.model_dump() == new.roots.model_dump()
|
|
assert captured.out == ""
|
|
assert captured.err == ""
|
|
assert len(warnings) == 1
|
|
|
|
|
|
def test_legacy_warning_does_not_contaminate_cli_json(tmp_path):
|
|
with pytest.warns(FutureWarning, match="DEPRECATION") as warnings:
|
|
result = CliRunner().invoke(
|
|
app,
|
|
["session", "list", "--json", "-c", str(_write_old_workspace(tmp_path))],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
json.loads(result.stdout)
|
|
assert "DEPRECATION" not in result.stdout
|
|
assert result.stderr == ""
|
|
assert len(warnings) == 1
|