120 lines
3.5 KiB
Python
120 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "plugins" / "1c"))
|
|
sys.path.insert(0, str(ROOT / "plugins" / "1c" / "connector"))
|
|
|
|
import adapter_1c_server # noqa: E402
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self) -> None:
|
|
self.rows = []
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def execute(self, query, _params=None) -> None:
|
|
if "FROM dbo.[Params]" in query:
|
|
self.rows = [
|
|
{"FileName": "a.ui", "PartNo": 0, "BinaryData": b"one"},
|
|
{"FileName": "a.ui", "PartNo": 1, "BinaryData": b"two"},
|
|
{"FileName": "b.si", "PartNo": 0, "BinaryData": b"three"},
|
|
]
|
|
else:
|
|
self.rows = []
|
|
|
|
def __iter__(self):
|
|
return iter(self.rows)
|
|
|
|
def fetchall(self):
|
|
return list(self.rows)
|
|
|
|
|
|
class FakeConnection:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def cursor(self, as_dict=False):
|
|
assert as_dict is True
|
|
return FakeCursor()
|
|
|
|
|
|
def test_repository_sql_state_snapshot_hashes_parts_without_payload(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
adapter_1c_server,
|
|
"connect_live_sql",
|
|
lambda base_id, method, timeout_seconds=120: (FakeConnection(), {"database": "db"}, None),
|
|
)
|
|
result = adapter_1c_server.repository_sql_state_snapshot(
|
|
{"base_id": "base", "tables": ["Params"], "include_extensions": False, "diagnostic": True}
|
|
)
|
|
assert result["status"] == "ok"
|
|
assert result["counts"] == {"records": 2, "extensions": 0}
|
|
first = result["records"][0]
|
|
assert first["file_name"] == "a.ui"
|
|
assert first["part_count"] == 2
|
|
assert first["bytes"] == 6
|
|
assert "payload" not in first
|
|
assert result["diagnostics"]["payload_returned"] is False
|
|
token_diff = adapter_1c_server.repository_sql_state_diff(
|
|
{
|
|
"before_token": result["comparison_token"],
|
|
"after_token": result["comparison_token"],
|
|
"diagnostic": True,
|
|
}
|
|
)
|
|
assert token_diff["status"] == "ok"
|
|
assert token_diff["changed"] is False
|
|
|
|
|
|
def test_repository_sql_state_diff_reports_changed_part() -> None:
|
|
before = {
|
|
"schema": "onec_repository_sql_state_snapshot.v1",
|
|
"base_id": "base",
|
|
"snapshot_id": "before",
|
|
"records": [
|
|
{
|
|
"table": "Params",
|
|
"file_name": "a.ui",
|
|
"bytes": 3,
|
|
"sha1": "old",
|
|
"part_count": 1,
|
|
"parts": [{"part_no": 0, "bytes": 3, "sha1": "old"}],
|
|
}
|
|
],
|
|
"extensions": [],
|
|
}
|
|
after = {
|
|
**before,
|
|
"snapshot_id": "after",
|
|
"records": [
|
|
{
|
|
"table": "Params",
|
|
"file_name": "a.ui",
|
|
"bytes": 4,
|
|
"sha1": "new",
|
|
"part_count": 1,
|
|
"parts": [{"part_no": 0, "bytes": 4, "sha1": "new"}],
|
|
}
|
|
],
|
|
}
|
|
result = adapter_1c_server.repository_sql_state_diff(
|
|
{"before": before, "after": after, "diagnostic": True}
|
|
)
|
|
assert result["status"] == "ok"
|
|
assert result["changed"] is True
|
|
assert result["counts"] == {"added": 0, "removed": 0, "modified": 1}
|
|
assert result["modified"][0]["key"] == "storage:Params:a.ui"
|
|
assert result["modified"][0]["changed_parts"] == ["0"]
|