149 lines
3.8 KiB
Python
149 lines
3.8 KiB
Python
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app
|
|
from tht.config import load_config
|
|
from tht.adapters.factory import build_vector_store
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_legacy_cli_subprocess_warns_once_on_stderr_and_keeps_json_stdout(tmp_path):
|
|
workspace = _write_old_workspace(tmp_path)
|
|
result = subprocess.run(
|
|
[
|
|
str(Path(__file__).parents[1] / ".venv" / "bin" / "tht"),
|
|
"session",
|
|
"list",
|
|
"--json",
|
|
"-c",
|
|
str(workspace),
|
|
],
|
|
cwd=tmp_path,
|
|
env={**os.environ, "PYTHONWARNINGS": "default"},
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
json.loads(result.stdout)
|
|
assert "DEPRECATION" not in result.stdout
|
|
assert result.stderr.count("DEPRECATION") == 1
|
|
|
|
|
|
def test_legacy_writer_only_vector_config_builds_for_targeted_writes(tmp_path):
|
|
workspace = _write_old_workspace(tmp_path)
|
|
content = workspace.read_text().replace(
|
|
"vector_rest:\n base_url: https://vectors.example.test/\n api_key: vector-reader\n",
|
|
"",
|
|
)
|
|
workspace.write_text(content)
|
|
|
|
with pytest.warns(FutureWarning):
|
|
cfg = load_config(workspace)
|
|
store = build_vector_store(cfg, require_write=True)
|
|
assert store.capabilities.search is False
|
|
assert store.capabilities.upsert is True
|