194 lines
7.6 KiB
Python
194 lines
7.6 KiB
Python
"""Pure regression fixtures for the proven ConfigInfo file-SHA1 map rewrite."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import sys
|
|
import zlib
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
CONNECTOR = ROOT / "plugins" / "1c" / "connector"
|
|
PARSER = ROOT / "plugins" / "1c"
|
|
for path in (CONNECTOR, PARSER):
|
|
if str(path) not in sys.path:
|
|
sys.path.insert(0, str(path))
|
|
|
|
import adapter_1c_server as adapter # noqa: E402
|
|
from parser.payload import decode_payload_lossless # noqa: E402
|
|
|
|
|
|
LOGICAL = "77494708-43ea-4956-ac3c-199cfb035ad2.0"
|
|
|
|
|
|
def raw_deflate(text: str) -> bytes:
|
|
encoder = zlib.compressobj(wbits=-zlib.MAX_WBITS)
|
|
return encoder.compress(text.encode("utf-8")) + encoder.flush()
|
|
|
|
|
|
def test_rewrites_only_named_file_sha1_reference() -> None:
|
|
original_file = b"old module bytes"
|
|
replacement_file = b"new module bytes"
|
|
old_ref = base64.b64encode(hashlib.sha1(original_file).digest()).decode("ascii")
|
|
service_atom = base64.b64encode(b"S" * 32).decode("ascii")
|
|
source_text = f'{{2,extension,{service_atom}}},{{7,"{LOGICAL}",{old_ref},"other.0",{old_ref}}}'
|
|
result = adapter.rewrite_extension_configinfo_file_sha1_reference(
|
|
raw_deflate(source_text), logical_file_name=LOGICAL, replacement_file_data=replacement_file
|
|
)
|
|
assert result["status"] == "ok"
|
|
assert result["replacement_file_sha1"] == hashlib.sha1(replacement_file).hexdigest()
|
|
decoded = str(decode_payload_lossless(result["payload"]).get("text") or "")
|
|
assert service_atom in decoded
|
|
assert result["replacement_reference_base64"] in decoded
|
|
assert decoded.count(old_ref) == 1 # unrelated map entry was preserved
|
|
|
|
|
|
def test_rejects_ambiguous_logical_reference() -> None:
|
|
ref = base64.b64encode(hashlib.sha1(b"old").digest()).decode("ascii")
|
|
source_text = f'{{7,"{LOGICAL}",{ref},"{LOGICAL}",{ref}}}'
|
|
result = adapter.rewrite_extension_configinfo_file_sha1_reference(
|
|
raw_deflate(source_text), logical_file_name=LOGICAL, replacement_file_data=b"new"
|
|
)
|
|
assert result["status"] == "reference_not_unique"
|
|
assert result["occurrences"] == 2
|
|
|
|
|
|
def test_builds_paired_plan_only_when_map_matches_current_stream() -> None:
|
|
current = b"current module"
|
|
replacement = b"replacement module"
|
|
current_ref = base64.b64encode(hashlib.sha1(current).digest()).decode("ascii")
|
|
configinfo = raw_deflate(f'{{7,"{LOGICAL}",{current_ref}}}')
|
|
result = adapter.build_extension_saved_state_pair_plan(
|
|
logical_file_name=LOGICAL,
|
|
current_file_data=current,
|
|
replacement_file_data=replacement,
|
|
current_configinfo_data=configinfo,
|
|
)
|
|
assert result["status"] == "ok"
|
|
assert result["module"]["replacement_sha1"] == hashlib.sha1(replacement).hexdigest()
|
|
assert result["configinfo"]["service_atoms"] == "preserved_unmodified"
|
|
|
|
|
|
def test_rejects_pair_when_map_points_to_another_stream() -> None:
|
|
mapped = b"other module"
|
|
current = b"current module"
|
|
mapped_ref = base64.b64encode(hashlib.sha1(mapped).digest()).decode("ascii")
|
|
configinfo = raw_deflate(f'{{7,"{LOGICAL}",{mapped_ref}}}')
|
|
result = adapter.build_extension_saved_state_pair_plan(
|
|
logical_file_name=LOGICAL,
|
|
current_file_data=current,
|
|
replacement_file_data=b"replacement",
|
|
current_configinfo_data=configinfo,
|
|
)
|
|
assert result["status"] == "configinfo_reference_mismatch"
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self, state: dict[str, bytes]) -> None:
|
|
self.state = state
|
|
self.rows: list[dict[str, object]] = []
|
|
|
|
def execute(self, sql: str, params: tuple[object, ...] | None = None) -> None:
|
|
if sql.startswith("SET LOCK_TIMEOUT"):
|
|
return
|
|
if sql.startswith("SELECT FileName, PartNo, BinaryData"):
|
|
assert params is not None
|
|
self.rows = [
|
|
{"FileName": name, "PartNo": 0, "BinaryData": self.state[name]}
|
|
for name in params
|
|
if name in self.state
|
|
]
|
|
return
|
|
if sql.startswith("UPDATE dbo.ConfigCASSave"):
|
|
assert params is not None
|
|
data, _size, name, _part = params
|
|
self.state[str(name)] = bytes(data)
|
|
return
|
|
raise AssertionError(sql)
|
|
|
|
def fetchall(self) -> list[dict[str, object]]:
|
|
return self.rows
|
|
|
|
|
|
class FakeConnection:
|
|
def __init__(self, state: dict[str, bytes]) -> None:
|
|
self.cursor_instance = FakeCursor(state)
|
|
self.committed = False
|
|
self.rolled_back = False
|
|
self.closed = False
|
|
|
|
def cursor(self, **_kwargs: object) -> FakeCursor:
|
|
return self.cursor_instance
|
|
|
|
def commit(self) -> None:
|
|
self.committed = True
|
|
|
|
def rollback(self) -> None:
|
|
self.rolled_back = True
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def install_fake_pair_sql(monkeypatch: object, state: dict[str, bytes], connection: FakeConnection) -> None:
|
|
monkeypatch.setattr(adapter, "connect_live_sql", lambda *_args, **_kwargs: (connection, {"database": "test"}, None))
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"read_storage_file_bytes",
|
|
lambda _base, _table, name, **_kwargs: (state.get(name), {"database": "test"}, None if name in state else {"status": "missing"}),
|
|
)
|
|
monkeypatch.setattr(adapter, "write_storage_apply_backup", lambda **_kwargs: {"backup_id": "test"})
|
|
monkeypatch.setattr(adapter, "invalidate_adapter_caches_after_saved_state_change", lambda *_args, **_kwargs: {"invalidated": True})
|
|
|
|
|
|
def test_atomically_updates_module_and_configinfo(monkeypatch: object) -> None:
|
|
extension = "fb26cf42-7609-11f1-828f-005056b0d483"
|
|
module_name = f"{extension}__{LOGICAL}"
|
|
configinfo_name = f"{extension}__configinfo"
|
|
current = b"current module"
|
|
current_ref = base64.b64encode(hashlib.sha1(current).digest()).decode("ascii")
|
|
state = {module_name: current, configinfo_name: raw_deflate(f'{{7,"{LOGICAL}",{current_ref}}}')}
|
|
connection = FakeConnection(state)
|
|
install_fake_pair_sql(monkeypatch, state, connection)
|
|
|
|
result = adapter.apply_extension_saved_state_pair(
|
|
"upo_test",
|
|
module_file_name=module_name,
|
|
replacement_module_data=b"new module",
|
|
expected_module_sha1=hashlib.sha1(current).hexdigest(),
|
|
proposal={"method": "test"},
|
|
)
|
|
|
|
assert result["status"] == "applied"
|
|
assert result["applied"] is True
|
|
assert connection.committed is True
|
|
assert result["configurator_refresh"]["action"] == "close_reopen_object"
|
|
map_text = str(decode_payload_lossless(state[configinfo_name]).get("text") or "")
|
|
assert base64.b64encode(hashlib.sha1(b"new module").digest()).decode("ascii") in map_text
|
|
|
|
|
|
def test_atomic_writer_rolls_back_when_configinfo_map_diverges(monkeypatch: object) -> None:
|
|
extension = "fb26cf42-7609-11f1-828f-005056b0d483"
|
|
module_name = f"{extension}__{LOGICAL}"
|
|
configinfo_name = f"{extension}__configinfo"
|
|
current = b"current module"
|
|
wrong_ref = base64.b64encode(hashlib.sha1(b"other module").digest()).decode("ascii")
|
|
state = {module_name: current, configinfo_name: raw_deflate(f'{{7,"{LOGICAL}",{wrong_ref}}}')}
|
|
connection = FakeConnection(state)
|
|
install_fake_pair_sql(monkeypatch, state, connection)
|
|
|
|
result = adapter.apply_extension_saved_state_pair(
|
|
"upo_test",
|
|
module_file_name=module_name,
|
|
replacement_module_data=b"new module",
|
|
expected_module_sha1=hashlib.sha1(current).hexdigest(),
|
|
proposal={"method": "test"},
|
|
)
|
|
|
|
assert result["status"] == "configinfo_reference_mismatch"
|
|
assert result["applied"] is False
|
|
assert connection.rolled_back is True
|
|
assert state[module_name] == current
|