241 lines
9.0 KiB
Python
241 lines
9.0 KiB
Python
"""Local mixed-owner admission/recovery evidence, not production capacity certification.
|
|
|
|
All inputs are synthetic and held in memory. The spawn observer temporarily
|
|
holds the admitted parent's handshake so the other owners encounter the same
|
|
occupied slot deterministically. Children perform real XLSX, template and
|
|
Dataflow work; there is no mocked process execution, database, or live service.
|
|
The non-queuing gate promises retryable rejection, not scheduler fairness.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from io import BytesIO
|
|
import json
|
|
import os
|
|
import threading
|
|
import time
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
try:
|
|
from openpyxl import Workbook
|
|
from govoplan_connectors.backend.tabular_adapters import (
|
|
parse_managed_tabular_content,
|
|
)
|
|
from govoplan_core.core.templates import TemplateRenderRequest
|
|
from govoplan_core.security import bounded_process
|
|
from govoplan_core.security.bounded_process import ProcessBudgetError
|
|
from govoplan_core.settings import settings
|
|
from govoplan_dataflow.backend.backends import execute_typed_graph
|
|
from govoplan_dataflow.backend.schemas import (
|
|
GraphEdge,
|
|
GraphNode,
|
|
GraphPosition,
|
|
PipelineGraph,
|
|
)
|
|
from govoplan_templates.backend.rendering import _render_payload
|
|
except ImportError as exc:
|
|
raise unittest.SkipTest(
|
|
"Mixed-owner isolation requires the optional module test environment."
|
|
) from exc
|
|
|
|
|
|
class IsolatedWorkCompositionTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
workbook = Workbook()
|
|
workbook.active.append(["name"])
|
|
workbook.active.append(["Ada"])
|
|
stream = BytesIO()
|
|
workbook.save(stream)
|
|
workbook.close()
|
|
self.workbook = stream.getvalue()
|
|
self.graph = PipelineGraph(
|
|
nodes=[
|
|
GraphNode(
|
|
id="source",
|
|
type="source.inline",
|
|
label="Source",
|
|
position=GraphPosition(x=0, y=0),
|
|
config={"source_name": "records", "rows": [{"name": "Ada"}]},
|
|
),
|
|
GraphNode(
|
|
id="output",
|
|
type="output",
|
|
label="Output",
|
|
position=GraphPosition(x=100, y=0),
|
|
config={},
|
|
),
|
|
],
|
|
edges=[GraphEdge(id="edge", source="source", target="output")],
|
|
)
|
|
|
|
def xlsx(self):
|
|
rows, sheet = parse_managed_tabular_content(
|
|
self.workbook,
|
|
filename="synthetic.xlsx",
|
|
content_type=None,
|
|
delimiter=",",
|
|
sheet_name=None,
|
|
)
|
|
self.assertEqual(rows, ({"name": "Ada"},))
|
|
self.assertEqual(sheet, "Sheet")
|
|
return "xlsx"
|
|
|
|
def templates(self):
|
|
payload, content_type, _pages = _render_payload(
|
|
SimpleNamespace(name="Synthetic template"),
|
|
SimpleNamespace(
|
|
content_text="Hello {{item.name}}",
|
|
content_html=None,
|
|
template_type="letter",
|
|
layout={},
|
|
output_profiles=[],
|
|
),
|
|
request=TemplateRenderRequest(
|
|
template_id="synthetic", output_format="text"
|
|
),
|
|
items=({"name": "Ada"},),
|
|
)
|
|
self.assertEqual(payload, b"Hello Ada")
|
|
self.assertEqual(content_type, "text/plain; charset=utf-8")
|
|
return "templates"
|
|
|
|
def dataflow(self):
|
|
result = execute_typed_graph(self.graph, backend="reference")
|
|
self.assertEqual(result.rows, [{"name": "Ada"}])
|
|
return "dataflow"
|
|
|
|
def test_one_shared_slot_rejects_other_owners_and_all_retries_recover(self):
|
|
owners = {
|
|
"xlsx": self.xlsx,
|
|
"templates": self.templates,
|
|
"dataflow": self.dataflow,
|
|
}
|
|
processes = []
|
|
modules = []
|
|
hold_next = False
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
observer_lock = threading.Lock()
|
|
maximum_unreaped = 0
|
|
observer_timeouts = 0
|
|
original_popen = bounded_process.subprocess.Popen
|
|
|
|
def observe_spawn(*args, **kwargs):
|
|
nonlocal hold_next, maximum_unreaped, observer_timeouts
|
|
process = original_popen(*args, **kwargs)
|
|
with observer_lock:
|
|
processes.append(process)
|
|
modules.append(args[0][5])
|
|
maximum_unreaped = max(
|
|
maximum_unreaped, sum(item.returncode is None for item in processes)
|
|
)
|
|
should_hold = hold_next
|
|
hold_next = False
|
|
if should_hold:
|
|
entered.set()
|
|
if not release.wait(8):
|
|
# Return control so the real runner's normal timeout and
|
|
# process-group cleanup still own this child on test error.
|
|
observer_timeouts += 1
|
|
return process
|
|
|
|
def rejected(operation):
|
|
try:
|
|
operation()
|
|
except Exception as exc:
|
|
cause = exc
|
|
while cause is not None and not isinstance(cause, ProcessBudgetError):
|
|
cause = cause.__cause__
|
|
self.assertIsInstance(cause, ProcessBudgetError)
|
|
self.assertEqual(cause.code, "busy")
|
|
return "busy"
|
|
self.fail(
|
|
"A different module admitted work while the shared slot was occupied."
|
|
)
|
|
|
|
started = time.monotonic()
|
|
busy_count = 0
|
|
try:
|
|
with (
|
|
patch.object(settings, "isolated_process_concurrency", 1),
|
|
patch.object(bounded_process.subprocess, "Popen", observe_spawn),
|
|
ThreadPoolExecutor(max_workers=3) as executor,
|
|
):
|
|
for owner, operation in owners.items():
|
|
with self.subTest(admitted_owner=owner):
|
|
entered.clear()
|
|
release.clear()
|
|
hold_next = True
|
|
holder = executor.submit(operation)
|
|
try:
|
|
self.assertTrue(
|
|
entered.wait(5),
|
|
"The admitted operation never spawned its real child.",
|
|
)
|
|
children_before = len(processes)
|
|
others = [
|
|
work for label, work in owners.items() if label != owner
|
|
]
|
|
denied = [
|
|
executor.submit(rejected, work) for work in others
|
|
]
|
|
self.assertEqual(
|
|
[future.result(timeout=5) for future in denied],
|
|
["busy", "busy"],
|
|
)
|
|
busy_count += len(denied)
|
|
self.assertEqual(len(processes), children_before)
|
|
finally:
|
|
release.set()
|
|
self.assertEqual(holder.result(timeout=15), owner)
|
|
self.assertEqual(bounded_process._active, 0)
|
|
# Every rejected owner is retried through its real API.
|
|
# Each must complete after the previous holder releases.
|
|
for other in others:
|
|
other()
|
|
self.assertEqual(bounded_process._active, 0)
|
|
finally:
|
|
release.set()
|
|
for process in processes:
|
|
self.assertIsNotNone(process.returncode, "Worker was not reaped.")
|
|
self.assertTrue(
|
|
all(
|
|
stream.closed
|
|
for stream in (process.stdin, process.stdout, process.stderr)
|
|
)
|
|
)
|
|
with self.assertRaises(ChildProcessError):
|
|
os.waitpid(process.pid, os.WNOHANG)
|
|
self.assertEqual(maximum_unreaped, 1)
|
|
self.assertEqual(observer_timeouts, 0)
|
|
self.assertEqual(len(processes), 9)
|
|
self.assertEqual(busy_count, 6)
|
|
self.assertEqual(
|
|
set(modules),
|
|
{
|
|
"govoplan_connectors.backend.tabular_adapters",
|
|
"govoplan_templates.backend.rendering",
|
|
"govoplan_dataflow.backend.backends.reference",
|
|
},
|
|
)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"local_composition": {
|
|
"successful_children": len(processes),
|
|
"busy_rejections": busy_count,
|
|
"maximum_unreaped_children": maximum_unreaped,
|
|
"all_children_reaped": True,
|
|
"seconds": round(time.monotonic() - started, 3),
|
|
}
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|