130 lines
5.7 KiB
Python
130 lines
5.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "plugins" / "1c" / "connector"))
|
|
|
|
import adapter_1c_server as adapter # noqa: E402
|
|
|
|
|
|
def test_sql_admin_save_load_and_redact_password(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
path = tmp_path / "bases.json"
|
|
monkeypatch.delenv("ONEC_SQL_BASES_JSON", raising=False)
|
|
monkeypatch.setenv("ONEC_SQL_BASES_JSON_FILE", str(path))
|
|
config = {"upo_test": {"server": "10.0.0.2", "database": "upo", "user": "reader", "password": "secret"}}
|
|
|
|
adapter.sql_admin_save(config)
|
|
|
|
assert adapter.sql_admin_load() == config
|
|
public = adapter.sql_admin_public(config)[0]
|
|
assert public["has_password"] is True
|
|
assert "password" not in public
|
|
assert "secret" not in json.dumps(public)
|
|
|
|
|
|
def test_sql_admin_rejects_inline_environment_config(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("ONEC_SQL_BASES_JSON", '{"base": {}}')
|
|
with pytest.raises(ValueError, match="ONEC_SQL_BASES_JSON_FILE"):
|
|
adapter.sql_admin_load()
|
|
|
|
|
|
@pytest.mark.parametrize("base_id", ["", "кириллица", "space id", "../escape"])
|
|
def test_sql_admin_rejects_unsafe_base_id(base_id: str) -> None:
|
|
with pytest.raises(ValueError, match="base_id"):
|
|
adapter.sql_admin_validate({"base_id": base_id, "server": "s", "database": "d", "user": "u"})
|
|
|
|
|
|
def test_sql_admin_accepts_password_environment_reference() -> None:
|
|
base_id, item = adapter.sql_admin_validate({
|
|
"base_id": "accounting-prod",
|
|
"server": "sql.internal",
|
|
"database": "Accounting",
|
|
"user": "reader",
|
|
"password": "ignored",
|
|
"password_env": "ONEC_SQL_PASSWORD_ACCOUNTING",
|
|
})
|
|
assert base_id == "accounting-prod"
|
|
assert item["password_env"] == "ONEC_SQL_PASSWORD_ACCOUNTING"
|
|
assert "password" not in item
|
|
|
|
|
|
def test_sql_admin_preserves_repository_user_from_settings() -> None:
|
|
_, item = adapter.sql_admin_validate({
|
|
"base_id": "upo_test",
|
|
"server": "sql.internal",
|
|
"database": "upo_test",
|
|
"user": "reader",
|
|
"repository": {
|
|
"backend": "karman_bridge",
|
|
"lock_mode": "manual",
|
|
"runner_url": "http://runner:8121",
|
|
"repository_user": "adm",
|
|
"repository_password_env": "REPO_PASSWORD",
|
|
"infobase_user": "designer-user",
|
|
"infobase_password_env": "IB_PASSWORD",
|
|
},
|
|
})
|
|
|
|
assert item["repository"]["repository_user"] == "adm"
|
|
assert item["repository"]["lock_mode"] == "manual"
|
|
assert item["repository"]["repository_password_env"] == "REPO_PASSWORD"
|
|
assert item["repository"]["infobase_user"] == "designer-user"
|
|
assert item["repository"]["infobase_password_env"] == "IB_PASSWORD"
|
|
|
|
|
|
def test_sql_admin_accepts_independent_base_and_extension_policies() -> None:
|
|
extension_guid = "11111111-2222-3333-4444-555555555555"
|
|
_, item = adapter.sql_admin_validate({
|
|
"base_id": "upo_test", "server": "sql", "database": "upo_test", "user": "reader",
|
|
"development_layers": {
|
|
"base": {"repository": {"mode": "manual"}, "support": {"mode": "rules", "rules": {"Документ.Заказ": "vendor_not_editable"}}},
|
|
f"extension:{extension_guid}": {"repository": {"mode": "none"}, "support": {"mode": "none"}},
|
|
},
|
|
})
|
|
assert item["development_layers"]["base"]["repository"]["mode"] == "manual"
|
|
assert item["development_layers"]["base"]["support"]["rules"]["Документ.Заказ"] == "vendor_not_editable"
|
|
assert item["development_layers"][f"extension:{extension_guid}"]["repository"]["mode"] == "none"
|
|
|
|
|
|
def test_sql_admin_rejects_extension_layer_without_guid() -> None:
|
|
with pytest.raises(ValueError, match="extension:<GUID>"):
|
|
adapter.sql_admin_validate({
|
|
"base_id": "upo_test", "server": "sql", "database": "upo_test", "user": "reader",
|
|
"development_layers": {"extension:test": {"repository": {"mode": "none"}, "support": {"mode": "none"}}},
|
|
})
|
|
|
|
|
|
def test_agent_can_persist_layer_repository_connection_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
path = tmp_path / "bases.json"
|
|
guid = "11111111-2222-3333-4444-555555555555"
|
|
monkeypatch.delenv("ONEC_SQL_BASES_JSON", raising=False)
|
|
monkeypatch.setenv("ONEC_SQL_BASES_JSON_FILE", str(path))
|
|
adapter.sql_admin_save({
|
|
"neft": {
|
|
"server": "sql", "database": "neft", "user": "reader",
|
|
"development_layers": {
|
|
"base": {"repository": {"mode": "manual", "connection_state": "configured"}, "support": {"mode": "unknown"}},
|
|
f"extension:{guid}": {"repository": {"mode": "unknown", "connection_state": "unknown"}, "support": {"mode": "unknown"}},
|
|
},
|
|
},
|
|
})
|
|
missing_confirmation = adapter.call_method("repository.layer.connection.set", {"base_id": "neft", "extension_guid": guid, "connection_state": "not_configured"})
|
|
assert missing_confirmation["status"] == "invalid_argument"
|
|
result = adapter.call_method("repository.layer.connection.set", {
|
|
"base_id": "neft", "extension_guid": guid, "connection_state": "not_configured", "repository_user": "repo-extension-user",
|
|
"confirm_repository_connection_change": True,
|
|
})
|
|
assert result["status"] == "saved"
|
|
assert result["repository_status"] == "repository_not_connected"
|
|
stored = adapter.sql_admin_load()
|
|
repository = stored["neft"]["development_layers"][f"extension:{guid}"]["repository"]
|
|
assert repository["connection_state"] == "not_configured"
|
|
assert repository["mode"] == "none"
|
|
assert repository["repository_user"] == "repo-extension-user"
|