diff --git a/.gitignore b/.gitignore index 2e051f4..7ee5269 100644 --- a/.gitignore +++ b/.gitignore @@ -166,3 +166,5 @@ test.pdf playwright/node_modules/ *.webm playwright/test-results/ + +coverage.lcov diff --git a/README.md b/README.md index 5f6eef8..272bd11 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ By leveraging TraceFlow, you can: - Maintain your documentation in the same git repository as your code, ensuring version control and easy collaboration. - Automatically produce a "validation pack" containing all requirements, design, and test plans in PDF format. - Generate a traceability matrix linking all requirements to all tests. +- Capture structured risk registers with requirement/test cross-links and colour-coded pre/post-mitigation risk levels. - Create fillable PDF forms for manual tests. - Run automated tests and capture their output as Markdown, then include them in the PDF report. - The result will look something [like this](example.pdf) @@ -25,6 +26,9 @@ project/ ├── design/ │ ├── design.md │ └── ... + ├── risks/ + │ ├── risk-register.md + │ └── ... └── tests/ ├── test_plan.md └── ... @@ -163,3 +167,33 @@ To address **REQ-003**, we will create a Python API for building and executing i - Functions to connect and execute pipeline components - Compliance checks to ensure the pipeline adheres to the required standards ``` + +### Risk Register Example +TraceFlow now understands risk registers and renders them on A3 landscape pages with colour-coded severity, probability, and residual risk ratings. + +``` +# Risk Register + +## RISK-001: Incorrect study-patient association + +Hazardous Situation: Clinician views wrong patient's images believing they are correct +Harm: Misdiagnosis, inappropriate treatment +Cause: Race condition during HL7/DICOM message processing leading to incorrect PatientID or AccessionNumber assignment +Severity: High +Probability: Medium +Controls: Unit/integration tests for identifier logic (REQ-005, TEST-002) +Residual Severity: Medium +Residual Probability: Low +Residual Risk: Operator review plus automatic quarantine when mismatches occur +``` + +### Generic Document Example +Any additional Markdown files (e.g., Installation/User Specifications) are included verbatim and can link to requirements, tests, or risks using their IDs: + +``` +# Installation & User Specification + +- Execute TEST-003 and record the release tag. +- Confirm the identifier reconciliation feature (REQ-005) is active before go-live. +- Review RISK-001 and RISK-002 residual risk statements with the clinical safety officer. +``` diff --git a/example.pdf b/example.pdf index afc8cc8..85f0712 100644 Binary files a/example.pdf and b/example.pdf differ diff --git a/examples/design/design-spec.md b/examples/design/design-spec.md index 365dd4a..a964d7e 100644 --- a/examples/design/design-spec.md +++ b/examples/design/design-spec.md @@ -1,25 +1,62 @@ # Functional Design Spec -## User Authentication +This document provides a high-level architectural view of the TraceFlow demo and links each design decision to the requirements, tests, and clinical safety risks that drive them. Refer to **RISK-001** when assessing identifier safety controls described below. -To address **REQ-001**, we will implement an authentication system using JWT (JSON Web Tokens). The system will include the following components: +## Architecture overview -- A login page with input fields for email and password -- A backend API endpoint to validate user credentials -- Middleware to validate JWT tokens for accessing protected resources +The system is split into ingestion, orchestration, and presentation tiers in order to satisfy **REQ-001** through **REQ-004** while maintaining a clean boundary for audit logging (**REQ-005**). -## MRI Dataset Import +```mermaid +graph TD + subgraph Ingestion + HL7[HL7 listener] + DICOM[DICOM listener] + end + subgraph Core Platform + Auth[Auth Service] + Reconcile[Identifier Reconciliation Engine] + Pipeline[Analysis Pipeline API] + Audit[Audit Ledger] + end + subgraph Presentation + UI[Clinician UI] + end + HL7 --> Reconcile + DICOM --> Reconcile + Reconcile --> Audit + Reconcile --> Pipeline + Auth --> UI + Pipeline --> UI + UI --> Audit +``` -To address **REQ-002**, we will develop a module to import MRI datasets in DICOM format. The module will include: +## Identifier reconciliation engine (REQ-005, RISK-001, TEST-002) -- A function to parse DICOM files -- Error handling for unsupported or malformed files -- Integration with the existing data storage system +The reconciliation engine protects against the hazardous situation captured in **RISK-001** where patient and study identifiers might be mismatched. Design goals: -## Image Analysis Pipeline +- Compare the DICOM `PatientID` and `AccessionNumber` with the HL7 metadata stream. +- If mismatches are detected, the transaction is quarantined and **TEST-002** exercises the operator override. +- Successful transactions emit structured audit events so **REQ-005** is met. -To address **REQ-003**, we will create a Python API for building and executing image analysis pipelines. This API will include: +### UML activity view -- A set of Python classes to represent pipeline components -- Functions to connect and execute pipeline components -- Compliance checks to ensure the pipeline adheres to the required standards \ No newline at end of file +1. Receive the incoming DICOM objects and perform schema checks. +2. Compare HL7 metadata with extracted tags and branch depending on whether identifiers match. +3. Persist successful studies and emit reconciliation metrics for **RISK-001**. +4. Quarantine mismatches and require an operator acknowledgement that is captured by **TEST-002**. + +## Authentication and authorization (REQ-001, TEST-001) + +Authentication relies on OpenID Connect so automated tests (**TEST-001**) can exercise both success and failure paths. Tokens carry the clinician's study permissions, and the middleware enforces them for every API endpoint. + +## Imaging pipeline (REQ-002, REQ-003, RISK-002) + +The MRI import module normalizes DICOM before invoking the analysis pipelines described in **REQ-003**. The pipelines emit validation artifacts: + +- DICOM specific logs for regulatory review. +- Image-derived measurements tied to their source requirement/test pairs. +- Alerts for saturation or out-of-range values, contributing to **RISK-002** mitigations. + +## Operational documents (General IUS) + +Operational guidance for installations is maintained in `docs/ius.md`. That supplemental document links directly to **TEST-003** for continuous integration evidence and references **RISK-002** to remind the operator of the residual hazards. diff --git a/examples/docs/ius.md b/examples/docs/ius.md new file mode 100644 index 0000000..b90c664 --- /dev/null +++ b/examples/docs/ius.md @@ -0,0 +1,28 @@ +# Installation & User Specification + +This generic operational document demonstrates how TraceFlow can include arbitrary Markdown artifacts that link back to requirements, tests, and risks. + +## Installation checklist + +1. Deploy the ingestion stack and confirm that the `traceflow` service exposes `/healthz`. +2. Verify that authentication (**REQ-001**) succeeds for a clinical account with the `clinician` role by following **TEST-001**. +3. Enable the identifier reconciliation engine (**REQ-005**) and record the resulting audit entry. +4. Capture Playwright evidence from **TEST-003** and attach it to the release ticket. + +## Operational monitoring + +- Operators review the reconciler dashboard hourly to ensure the guardrails for **RISK-001** and **RISK-002** remain in place. +- When a mismatch occurs, follow the escalation steps documented in the clinical SOP and reference the impacted **REQ-005** controls. +- Residual risks shall be re-evaluated whenever a new modality is onboarded. + +## Traceability table + +| Step | Linked Requirement/Test | Linked Risk | +|-----------------------------------------|-------------------------|-------------| +| Configure audit webhooks | REQ-005 | RISK-001 | +| Re-run integration suite after upgrade | TEST-003 | RISK-002 | +| Attach validation pack to submission | TEST-001 / TEST-002 | RISK-003 | + +## Sign-off + +The implementation owner confirms that installation and operational verification were executed per procedure and that the TraceFlow validation pack is archived with the evidence references listed above. diff --git a/examples/requirements/basic-requirements.md b/examples/requirements/basic-requirements.md index fb99912..1cf734e 100644 --- a/examples/requirements/basic-requirements.md +++ b/examples/requirements/basic-requirements.md @@ -14,7 +14,7 @@ $$ H = L \times \log_2(N) $$ Where $H$ is the entropy, $L$ is the password length, and $N$ is the number of possible symbols. -### Example inline code formatting +### Example inline code formatting The ID of the user is, e.g., `1af345e6`. @@ -47,15 +47,39 @@ def hello_world(): print("Hello world!") ``` -## REQ-004: Output data +## REQ-004: Output data The platform must be able to export the results of image analysis pipelines in a standard format, including the following parameters: | **DICOM Series** | **Key Parameters** | **Typical Parameters** | **Map to Image Type** | |------------------------|---------------------------------|---------------------------------------------------------------|-----------------------| -| T1 VFA | TR, TE, FA | 96x96x24

FA=2°,17°,32° | `vfa` | -| High-res pre-contrast | TR, TE, FA | 512x512x92

FA=32° | `high-res-pre` | -| T1-weighted dynamic | TR, TE, FA, Temporal resolution | 96x96x24

FA=17°

Temporal resolution: 2s to 10s | `dynamic-uncorrected` | -| High-res post-contrast | TR, TE, FA | 512x512x92

FA=32° | `high-res-post` | +| T1 VFA | TR, TE, FA | 96x96x24 FA=2°,17°,32° | `vfa` | +| High-res pre-contrast | TR, TE, FA | 512x512x92 FA=32° | `high-res-pre` | +| T1-weighted dynamic | TR, TE, FA, Temporal resolution | 96x96x24 FA=17° Temporal resolution: 2s to 10s | `dynamic-uncorrected` | +| High-res post-contrast | TR, TE, FA | 512x512x92 FA=32° | `high-res-post` | +## REQ-005: Patient-study reconciliation + +The ingestion service shall ensure that study metadata stays associated with the correct patient identifier to reduce **RISK-001** and **RISK-002**. + +- Cross-validate the incoming HL7 and DICOM identifiers before persisting them. +- Flag mismatches for operator review and capture evidence via TEST-002. +- Provide a reconciliation webhook so downstream systems can resynchronize identifiers. + +### Example sequence diagram + +```mermaid +sequenceDiagram + participant PACS + participant TraceFlow + participant Operator + PACS->>TraceFlow: DICOM C-STORE (StudyUID=123) + TraceFlow->>TraceFlow: Validate identifiers (REQ-005) + alt Identifier mismatch + TraceFlow->>Operator: Alert referencing TEST-002 + Operator->>TraceFlow: Accept or reject mapping + else Identifiers aligned + TraceFlow-->>PACS: ACK with audit log entry + end +``` diff --git a/examples/risks/risk-register.md b/examples/risks/risk-register.md new file mode 100644 index 0000000..dba0c94 --- /dev/null +++ b/examples/risks/risk-register.md @@ -0,0 +1,223 @@ +# Risk Register + +TraceFlow libraries track software hazards alongside the requirements and tests that mitigate them. + +## RISK-001: Incorrect study-patient association + +Hazardous Situation: Clinician views the wrong patient's images believing they are correct. + +Harm: Misdiagnosis or inappropriate treatment. + +Cause: Race condition during HL7/DICOM message processing leading to incorrect `PatientID` or `AccessionNumber`. + +Severity: High + +Probability: Medium + +Controls: Identifier reconciliation service (**REQ-005**) exercised by **TEST-002**. + +Residual Severity: Medium + +Residual Probability: Low + +Residual Risk: Operator review plus automatic quarantine when mismatches occur. + +## RISK-002: Undetected imaging artifacts + +Hazardous Situation: Poor quality MRI acquisitions flow through the analysis pipeline without alerts. + +Harm: False clinical conclusions or delayed treatment. + +Cause: Lack of automated QA gates between the import module (**REQ-002**) and analysis pipeline (**REQ-003**). + +Severity: Medium + +Probability: Medium + +Controls: Automated test matrix (**TEST-003**) and image QC hooks inside the pipeline. + +Residual Severity: Low + +Residual Probability: Low + +Residual Risk: Clinician-facing dashboard highlights residual warnings for manual acknowledgement. + +## RISK-003: Audit trail corruption + +Hazardous Situation: Investigators cannot reconstruct prior runs because audit entries were dropped. + +Harm: Compliance breach or inability to support safety investigations. + +Cause: Misconfigured storage or missing webhook callbacks when persisting audit artifacts from **REQ-005**. + +Severity: Medium + +Probability: Low + +Controls: Continuous integration checks (**TEST-003**) and deployment checklist in `docs/ius.md`. + +Residual Severity: Low + +Residual Probability: Low + +Residual Risk: Acceptable provided long-term storage health checks pass weekly. + +## RISK-004: Unauthorized workspace access + +Hazardous Situation: An unapproved clinician launches the UI and edits live studies without authentication. + +Harm: Breach of patient confidentiality and risk of tampering with diagnostic evidence. + +Cause: Misconfigured identity provider or bypass of the login screen covered in **REQ-001**. + +Severity: High + +Probability: Low + +Controls: OpenID Connect gateway, MFA, and **TEST-001** regression runs. + +Residual Severity: Medium + +Residual Probability: Low + +Residual Risk: Acceptable when the security team performs quarterly access recertification. + +## RISK-005: Notification backlog for identifier mismatches + +Hazardous Situation: Identifier reconciliation alerts pile up, so clinicians miss urgent mismatches. + +Harm: Delayed detection of study/patient mix-ups (see **REQ-005**). + +Cause: Downstream webhook queue saturation or operator console outages. + +Severity: Medium + +Probability: Medium + +Controls: Autoscaling webhook workers plus the workflow exercised by **TEST-002**. + +Residual Severity: Low + +Residual Probability: Medium + +Residual Risk: Escalate to the on-call operator when more than five alerts remain untriaged for over 30 minutes. + +## RISK-006: Exported report mismatch + +Hazardous Situation: Generated PDF or DICOM SR exports omit the source dataset IDs. + +Harm: Inability to justify clinical findings or to reconstruct the original evidence trail. + +Cause: Faulty serialization of the output schema defined in **REQ-004**. + +Severity: Medium + +Probability: Low + +Controls: Automated CI pipeline (**TEST-003**) that validates exports against golden files. + +Residual Severity: Low + +Residual Probability: Low + +Residual Risk: Acceptable when QA signs off release bundles that include checksum manifests. + +## RISK-007: Pipeline drift between releases + +Hazardous Situation: Model weights or preprocessing steps change without validation, yielding inconsistent diagnoses. + +Harm: False positives/negatives that could trigger incorrect therapy. + +Cause: Manual edits to **REQ-003** pipeline components without rerunning validation. + +Severity: High + +Probability: Medium + +Controls: Continuous integration gates in **TEST-003** plus code reviews for every pipeline update. + +Residual Severity: Medium + +Residual Probability: Low + +Residual Risk: Acceptable once deployment reports attach the validation pack and a signed change record. + +## RISK-008: Data retention policy breach + +Hazardous Situation: Backups omit audit logs, preventing reconstruction of historic identifier fixes. + +Harm: Compliance violations with ICH-GCP and inability to support investigations. + +Cause: Retention scripts ignore the storage bucket defined in **REQ-005**. + +Severity: Medium + +Probability: Medium + +Controls: Scheduled integrity checks plus **TEST-003** to validate log export scripts. + +Residual Severity: Low + +Residual Probability: Low + +Residual Risk: Acceptable when quarterly SOP reviews confirm retention evidence is archived. + +## RISK-009: Automated test evidence missing + +Hazardous Situation: Regression runs referenced by **TEST-003** fail but their output is not included in the validation pack. + +Harm: Release approvals rely on incomplete evidence trails. + +Cause: Operator forgets to pass `--playwright-dir` or the CI job skips the autotest block. + +Severity: Medium + +Probability: Medium + +Controls: TraceFlow CLI guards that raise on missing artifacts plus the process captured in `docs/ius.md`. + +Residual Severity: Low + +Residual Probability: Low + +Residual Risk: Acceptable provided release managers verify example.pdf before sign-off. + +## RISK-010: External interface misconfiguration + +Hazardous Situation: Downstream systems consume reconciliation webhooks with the wrong schema and silently drop events. + +Harm: Identifier mismatches never reach clinical users even though TraceFlow generated the alerts. + +Cause: Integration guides diverge from the specification in **REQ-005** or design doc steps. + +Severity: Medium + +Probability: Medium + +Controls: Installation & User Specification walkthrough (`docs/ius.md`) and contract tests executed as part of **TEST-002**. + +Residual Severity: Low + +Residual Probability: Medium + +Residual Risk: Acceptable once each release captures signed integration test outputs in the validation pack. + +## RISK-011: Manual test paperwork incomplete + +Hazardous Situation: Manual acceptance tests (**TEST-001**) are executed but signatures or pass/fail states are unsigned, invalidating the safety case. + +Harm: Regulatory submissions may be rejected, delaying patient access. + +Cause: Clinicians forget to fill the interactive PDF fields exported by TraceFlow. + +Severity: Low + +Probability: Medium + +Controls: The test cover sheet plus electronic signature workflow baked into TraceFlow forms. + +Residual Severity: Low + +Residual Probability: Low + +Residual Risk: Acceptable when QA audits confirm PDF form completion before release. diff --git a/examples/tests/manual-acceptance-test.md b/examples/tests/manual-acceptance-test.md index 1019ddb..cc4c5cc 100644 --- a/examples/tests/manual-acceptance-test.md +++ b/examples/tests/manual-acceptance-test.md @@ -28,10 +28,12 @@ The user is logged in and redirected to the main dashboard. ## TEST-002: Automatic Playwright test **Requirement ID:** REQ-003 +**Requirement ID:** REQ-005 ### Test Steps: 1. The `autoplaywright` block runs the Playwright sample located in the `playwright/` directory and captures the video, stdout/stderr, and a 3×3 key frame grid from the test execution. +2. The scripted run intentionally drives a patient-identifier edit workflow to demonstrate the mitigation for **RISK-001**. ### Expected Result: @@ -50,6 +52,7 @@ The test should pass successfully. ### Test Steps: 1. This `autotest` block demonstrates running a simple pytest command as part of the report. +2. The quick-running suite doubles as a guardrail for **RISK-002** and **RISK-003**; failures block deployments. ### Expected Result: diff --git a/tests/test_parser.py b/tests/test_parser.py index e69de29..8a062ba 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +import textwrap +from pathlib import Path + +import pytest + +from traceflow.parser import process_directory + + +def _write_markdown(path: Path, content: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + normalised = textwrap.dedent(content).strip() + path.write_text(normalised + "\n") + + +def test_process_directory_parses_risks_and_design(tmp_path: Path) -> None: + requirements_path = Path(tmp_path) / "requirements" / "requirements.md" + tests_path = Path(tmp_path) / "tests" / "tests.md" + risks_path = Path(tmp_path) / "risks" / "risk-register.md" + design_path = Path(tmp_path) / "design" / "design-overview.md" + docs_path = Path(tmp_path) / "docs" / "ius.md" + + _write_markdown( + requirements_path, + """ + # Requirements + + ## REQ-001: Login + + The system shall authenticate clinicians. + """, + ) + + _write_markdown( + tests_path, + """ + # Tests + + ## TEST-001: Login succeeds + + **Requirement ID:** REQ-001 + """, + ) + + _write_markdown( + risks_path, + """ + # Risk Register + + Intro paragraph. + + ## RISK-001: Incorrect association + + Hazardous Situation: Images are shown for the wrong patient + + Harm: Misdiagnosis + + Cause: Race condition during message ingestion + + Severity: High + + Probability: Medium + + Controls: Requirements REQ-001 verified by TEST-001 + + Residual Severity: Medium + + Residual Probability: Low + """, + ) + + _write_markdown( + design_path, + """ + # Design Overview + + ## Authentication + + Refers to REQ-001. + """, + ) + + _write_markdown( + docs_path, + """ + # Installation Qualification + + Ensure TEST-001 evidence is archived. + """, + ) + + document = process_directory(str(tmp_path), version="1.0.0") + + assert len(document.risks) == 1 + risk_doc = document.risks[0] + assert risk_doc.title == "Risk Register" + assert len(risk_doc.items) == 1 + risk = risk_doc.items[0] + assert risk.risk_id == "RISK-001" + assert risk.attributes["severity"] == "High" + assert risk.attributes["probability"] == "Medium" + assert risk.attributes["residual_severity"] == "Medium" + assert sorted(risk.requirement_refs) == ["REQ-001"] + assert sorted(risk.test_refs) == ["TEST-001"] + + assert len(document.design_documents) == 1 + assert document.design_documents[0].category == "design" + assert len(document.supplementary_documents) == 1 + assert document.supplementary_documents[0].category == "general" + + +def test_risk_reference_validation(tmp_path: Path) -> None: + requirements_path = Path(tmp_path) / "requirements" / "requirements.md" + risks_path = Path(tmp_path) / "risks" / "risk-register.md" + + _write_markdown( + requirements_path, + """ + # Requirements + + ## REQ-100: Example + + Some body text. + """, + ) + + _write_markdown( + risks_path, + """ + # Risk Register + + ## RISK-404: Missing control + + Controls: Mitigated by REQ-999 + """, + ) + + with pytest.raises(ValueError, match="Risk `RISK-404` references requirement `REQ-999`"): + process_directory(str(tmp_path), version="0.1.0") diff --git a/tests/test_pdf_generator.py b/tests/test_pdf_generator.py index 363460f..3ac10a6 100644 --- a/tests/test_pdf_generator.py +++ b/tests/test_pdf_generator.py @@ -8,3 +8,13 @@ def test_process_text(self) -> None: latex = PdfReport.process_text_impl(text, {"UNIQUE-ID-001"}) expected = r"This is some text with a \hyperref[UNIQUE-ID-001]{UNIQUE-ID-001} and \texttt{code formatted} text too." # noqa self.assertEqual(latex, expected) + + def test_evaluate_risk_rating(self) -> None: + label, score, colour = PdfReport._evaluate_risk_rating("High", "Medium") + self.assertEqual(label, "High") + self.assertEqual(score, 12) + self.assertTrue(colour.startswith("orange")) + + label, score, colour = PdfReport._evaluate_risk_rating("1", "2") + self.assertEqual(label, "Low") + self.assertEqual(score, 2) diff --git a/traceflow/parser.py b/traceflow/parser.py index 33806df..39b47aa 100644 --- a/traceflow/parser.py +++ b/traceflow/parser.py @@ -1,6 +1,8 @@ import os -from typing import Union, List, Generic, TypeVar, Callable, Type +import re from dataclasses import dataclass +from pathlib import Path +from typing import Union, List, Generic, TypeVar, Callable, Type import mistune import yaml @@ -23,21 +25,31 @@ class Test: @dataclass -class Design: - design_id: str +class Risk: + risk_id: str content: list[dict] title: str + attributes: dict[str, str] + requirement_refs: list[str] + test_refs: list[str] - def get_referenced_requirement_ids(self) -> list[str]: - """ A list of requirement IDs that are present in the description """ - raise NotImplementedError - def get_referenced_test_ids(self) -> list[str]: - """ A list of test IDs that are present in the description """ - raise NotImplementedError +@dataclass +class MarkdownDocument: + title: str + filename: str + content: list[dict] + category: str + + @staticmethod + def from_file(file_path: str, category: str) -> 'MarkdownDocument': + parsed = parse_markdown(read_file(file_path)) + title = extract_title(parsed) + content = parsed[1:] + return MarkdownDocument(title=title, filename=file_path, content=content, category=category) -T = TypeVar('T', bound=Union['Requirement', 'Test']) +T = TypeVar('T', bound=Union['Requirement', 'Test', 'Risk']) C = TypeVar('C', bound='SubDocument') @@ -137,10 +149,142 @@ def from_file(file_path: str) -> 'TestDocument': return SubDocument.from_file_impl(TestDocument, file_path, TestDocument.item_generator) +class RiskDocument(SubDocument[Risk]): + + _FIELD_ALIASES: dict[str, str] = { + "hazardous situation": "hazardous_situation", + "hazard": "hazardous_situation", + "harm": "harm", + "cause": "cause", + "severity": "severity", + "probability": "probability", + "risk estimate": "risk_estimate", + "risk estimation": "risk_estimate", + "controls": "controls", + "risk control": "controls", + "risk controls": "controls", + "risk control measure": "controls", + "residual risk": "residual_risk", + "residual risk assessment": "residual_risk", + "residual severity": "residual_severity", + "residual probability": "residual_probability", + "detection": "detection", + "linked requirement": "linked_requirements", + "linked requirements": "linked_requirements", + "linked test": "linked_tests", + "linked tests": "linked_tests", + } + + _REQ_PATTERN = re.compile(r"\bREQ-[A-Za-z0-9_-]+\b", re.IGNORECASE) + _TEST_PATTERN = re.compile(r"\bTEST-[A-Za-z0-9_-]+\b", re.IGNORECASE) + + @staticmethod + def _normalise_field_name(raw: str) -> str | None: + normalised = re.sub(r"[^a-z0-9 ]", "", raw.lower().strip()) + return RiskDocument._FIELD_ALIASES.get(normalised) + + @staticmethod + def _flatten_text(elem: dict) -> str: + if elem["type"] == "text": + return elem.get("text", "") + if elem["type"] == "codespan": + return elem.get("text", "") + + if elem["type"] in {"softbreak", "linebreak"}: + return "\n" + + text = "" + for child in elem.get("children", []): + text += RiskDocument._flatten_text(child) + return text + + @staticmethod + def _extract_field(elem: dict) -> tuple[str, str] | None: + if elem["type"] != "paragraph": + return None + text = RiskDocument._flatten_text(elem).strip() + if ":" not in text: + return None + field_name, value = text.split(":", 1) + canonical = RiskDocument._normalise_field_name(field_name) + if canonical is None: + return None + return canonical, value.strip() + + @staticmethod + def _extract_references(text: str) -> tuple[list[str], list[str]]: + reqs = [match.upper() for match in RiskDocument._REQ_PATTERN.findall(text)] + tests = [match.upper() for match in RiskDocument._TEST_PATTERN.findall(text)] + return reqs, tests + + @staticmethod + def item_generator(parsed_content: list[dict]) -> list[Risk]: + risks: list[Risk] = [] + current_risk = Risk( + risk_id="", + content=[], + title="", + attributes={}, + requirement_refs=[], + test_refs=[], + ) + for elem in parsed_content: + if is_ast_element_heading(elem) == 2: + if len(current_risk.content) > 0: + risks.append(current_risk) + current_risk = Risk( + risk_id="", + content=[], + title="", + attributes={}, + requirement_refs=[], + test_refs=[], + ) + heading_text = get_heading_text(elem) + current_risk.risk_id = heading_text.split(" ")[0].replace(":", "") + current_risk.title = heading_text.replace(current_risk.risk_id + ":", "").strip() + else: + if current_risk.risk_id != "": + current_risk.content.append(elem) + extracted_field = RiskDocument._extract_field(elem) + flattened = RiskDocument._flatten_text(elem) + req_refs, test_refs = RiskDocument._extract_references(flattened) + for req in req_refs: + if req not in current_risk.requirement_refs: + current_risk.requirement_refs.append(req) + for test in test_refs: + if test not in current_risk.test_refs: + current_risk.test_refs.append(test) + if extracted_field: + field_name, field_value = extracted_field + current_risk.attributes[field_name] = field_value + # If the field explicitly lists linked requirements/tests, merge them back in + if field_name in {"linked_requirements", "controls"}: + reqs_in_field, _ = RiskDocument._extract_references(field_value) + for req in reqs_in_field: + if req not in current_risk.requirement_refs: + current_risk.requirement_refs.append(req) + if field_name in {"linked_tests", "controls"}: + _, tests_in_field = RiskDocument._extract_references(field_value) + for test in tests_in_field: + if test not in current_risk.test_refs: + current_risk.test_refs.append(test) + if len(current_risk.content) > 0: + risks.append(current_risk) + return risks + + @staticmethod + def from_file(file_path: str) -> 'RiskDocument': + return SubDocument.from_file_impl(RiskDocument, file_path, RiskDocument.item_generator) + + @dataclass class Document: requirements: list[RequirementDocument] tests: list[TestDocument] + risks: list[RiskDocument] + design_documents: list[MarkdownDocument] + supplementary_documents: list[MarkdownDocument] name: str = "" input_dir: str = "" version: str = "" @@ -150,6 +294,7 @@ def verify_all_ids_unique(self) -> None: all_ids: list[str] = ( [req.req_id for r in self.requirements for req in r.items] + [test.test_id for t in self.tests for test in t.items] + + [risk.risk_id for risk_doc in self.risks for risk in risk_doc.items] ) if len(all_ids) != len(set(all_ids)): @@ -176,9 +321,25 @@ def build_traceability_matrices(self) -> None: if req_id not in all_requirement_ids: raise ValueError(f"Test `{t.test_id}` references requirement `{req_id}` which does not exist") + def verify_risk_references(self) -> None: + """Ensure that any requirement/test IDs referenced within the risk register exist.""" + known_requirements: set[str] = {item.req_id for doc in self.requirements for item in doc.items} + known_tests: set[str] = {item.test_id for doc in self.tests for item in doc.items} + for risk_doc in self.risks: + for risk in risk_doc.items: + for req_id in risk.requirement_refs: + if req_id not in known_requirements: + raise ValueError( + f"Risk `{risk.risk_id}` references requirement `{req_id}` which does not exist" + ) + for test_id in risk.test_refs: + if test_id not in known_tests: + raise ValueError(f"Risk `{risk.risk_id}` references test `{test_id}` which does not exist") + def __post_init__(self) -> None: self.verify_all_ids_unique() self.build_traceability_matrices() + self.verify_risk_references() def read_file(file_path: str) -> str: @@ -194,22 +355,33 @@ def parse_markdown(content: str) -> list[dict]: def process_directory(directory: str, version: str) -> Document: requirements = [] tests = [] + risks = [] + design_documents: list[MarkdownDocument] = [] + supplementary_documents: list[MarkdownDocument] = [] for root, _, files in os.walk(directory): for file in files: if file.endswith(".md"): file_path = os.path.join(root, file) - if "req" in file_path.lower(): + relative_path = os.path.relpath(file_path, directory) + path_components = [part.lower() for part in Path(relative_path).parts] + top_component = path_components[0] + filename_component = path_components[-1] + + if top_component.startswith("req") or filename_component.startswith("req"): requirements.append(RequirementDocument.from_file(file_path)) - elif "test" in file_path.lower(): + elif top_component.startswith("test") or filename_component.startswith("test"): tests.append(TestDocument.from_file(file_path)) - elif "design" in file_path.lower(): - print("Warning, skipping design parsing (not implemented)") + elif ( + top_component.startswith("risk") + or filename_component.startswith("risk") + or top_component.startswith("hazard") + or filename_component.startswith("hazard") + ): + risks.append(RiskDocument.from_file(file_path)) + elif top_component.startswith("design") or filename_component.startswith("design"): + design_documents.append(MarkdownDocument.from_file(file_path, category="design")) else: - print( - f"Directory is: {directory}, unsure what to parse as. Directory should contain" - " `requirements`, `tests` or `design`" - ) - continue + supplementary_documents.append(MarkdownDocument.from_file(file_path, category="general")) absolute_dir_path = os.path.abspath(directory) # Check if config.yml exists in the directory @@ -220,7 +392,16 @@ def process_directory(directory: str, version: str) -> Document: config = yaml.safe_load(config_file) if "name" in config: name = config["name"] - return Document(requirements=requirements, tests=tests, name=name, input_dir=absolute_dir_path, version=version) + return Document( + requirements=requirements, + tests=tests, + risks=risks, + design_documents=design_documents, + supplementary_documents=supplementary_documents, + name=name, + input_dir=absolute_dir_path, + version=version, + ) def is_ast_element_heading(elem: dict) -> int: diff --git a/traceflow/pdf_generator.py b/traceflow/pdf_generator.py index f47490e..204fca1 100644 --- a/traceflow/pdf_generator.py +++ b/traceflow/pdf_generator.py @@ -20,7 +20,13 @@ else: _CAIROSVG_IMPORT_ERROR = None -from traceflow.parser import Document, RequirementDocument, parse_markdown +from traceflow.parser import ( + Document, + MarkdownDocument, + RequirementDocument, + RiskDocument, + parse_markdown, +) from traceflow.version import __version__ _latex_jinja2_env = latex.jinja2.make_env() @@ -71,6 +77,36 @@ def isolated_filesystem(temp_path: Optional[str] = None) -> Generator: class PdfReport(): + _RISK_SCALE: dict[str, int] = { + "very high": 5, + "critical": 5, + "catastrophic": 5, + "extreme": 5, + "high": 4, + "major": 4, + "serious": 4, + "frequent": 4, + "medium": 3, + "moderate": 3, + "occasional": 3, + "possible": 3, + "low": 1, + "minor": 1, + "remote": 1, + "unlikely": 1, + "very low": 1, + "negligible": 1, + "rare": 1, + "improbable": 1, + } + + _RISK_LEVELS: list[tuple[int, str, str]] = [ + (16, "Critical", "red!60"), + (9, "High", "orange!65"), + (4, "Medium", "yellow!40"), + (1, "Low", "green!35"), + ] + @staticmethod def process_text_impl(text: str, unique_ids: set[str]) -> str: @@ -80,9 +116,14 @@ def process_text_impl(text: str, unique_ids: set[str]) -> str: # 2. For each word, check if it is a unique ID. unique_ids is a set, so this is O(1) for index, word in enumerate(words): - altered_word = word.replace(":", "") - if altered_word in unique_ids: - words[index] = f"\\hyperref[{altered_word}]{{{word}}}" + leading = len(word) - len(word.lstrip("()[]{}.,;:<>")) + trailing = len(word.rstrip("()[]{}.,;:<>")) + prefix = word[:leading] + suffix = word[trailing:] + core = word[leading:trailing] if trailing > leading else word[leading:] + altered_word = core.replace(":", "") + if altered_word in unique_ids and core: + words[index] = f"{prefix}\\hyperref[{altered_word}]{{{core}}}{suffix}" # 3. Rebuild the text new_text = " ".join(words) @@ -108,6 +149,46 @@ def process_text_impl(text: str, unique_ids: set[str]) -> str: def process_text(self, text: str) -> str: return self.process_text_impl(text, self.unique_ids) + @staticmethod + def _build_label_from_text(prefix: str, text: str) -> str: + base = re.sub(r"[^A-Za-z0-9]+", "-", text).strip("-").lower() + if not base: + base = "doc" + return f"{prefix}-{base}" + + @classmethod + def _score_risk_dimension(cls: type['PdfReport'], value: str) -> int: + if not value: + return 0 + digits = re.findall(r"\d+", value) + if digits: + try: + return int(digits[0]) + except ValueError: + pass + normalised = re.sub(r"[^a-z ]", "", value.lower()).strip() + return cls._RISK_SCALE.get(normalised, 0) + + @classmethod + def _evaluate_risk_rating(cls: type['PdfReport'], severity: str, probability: str) -> tuple[str, int, str]: + severity_score = cls._score_risk_dimension(severity) + probability_score = cls._score_risk_dimension(probability) + score = severity_score * probability_score + if score == 0: + return "", 0, "" + for threshold, label, colour in cls._RISK_LEVELS: + if score >= threshold: + return label, score, colour + return "", score, "" + + def _format_risk_rating_cell(self, label: str, score: int, colour: str) -> str: + if not label: + return "" + cell_text = self.process_text(f"{label} ({score})") + if colour: + return f"\\cellcolor{{{colour}}}{cell_text}" + return cell_text + def build_traceability_matrix(self, req_page: RequirementDocument) -> str: # Display the traceability matrix table = "\\subsection{Traceability Matrix}\n\n" @@ -174,6 +255,149 @@ def build_traceability_matrix(self, req_page: RequirementDocument) -> str: return table + def render_markdown_document(self, doc: MarkdownDocument) -> str: + heading = doc.title + if doc.category == "design": + heading = f"Design - {doc.title}" + elif doc.category not in {"general", "design"}: + heading = f"{doc.category.title()} - {doc.title}" + label = self._build_label_from_text(doc.category or "doc", doc.title) + latex = "\\section{" + self.process_text(heading) + "}\\label{" + label + "}\n\n" + latex += self.md_to_latex(doc.content) + latex += "\n\n\\newpage\n\n" + return latex + + def build_risk_register(self, risk_page: RiskDocument) -> str: + if not risk_page.items: + return "" + + column_fragments = [ + "@{}p{4.0cm}", + "p{5.0cm}", + "p{4.2cm}", + "p{4.2cm}", + "p{4.2cm}", + "p{6.3cm}", + "p{5.5cm}", + "@{}", + ] + column_spec = "".join(column_fragments) + controls_width = 6.3 + residual_width = 5.5 + + table_lines = [ + "\\clearpage", + "\\begingroup", + "\\setlength{\\paperwidth}{420mm}", + "\\setlength{\\paperheight}{297mm}", + "\\pdfpagewidth=420mm", + "\\pdfpageheight=297mm", + "\\special{papersize=420mm,297mm}", + "\\newgeometry{paperwidth=420mm,paperheight=297mm,left=15mm,right=15mm,top=20mm,bottom=20mm}", + "\\footnotesize", + "\\setlength\\tabcolsep{3pt}", + "\\renewcommand{\\arraystretch}{1.3}", + "\\setlength\\LTleft{0pt}", + "\\setlength\\LTright{0pt}", + f"\\begin{{longtable}}{{{column_spec}}}", + ] + + header_cells = [ + "\\textbf{Risk ID}", + "\\textbf{Hazardous Situation}", + "\\textbf{Harm}", + "\\textbf{Cause}", + "\\textbf{Risk (Severity $\\times$ Probability)}", + "\\textbf{Controls}", + "\\textbf{Residual Risk}", + ] + header_row = " & ".join(header_cells) + " \\\\ \\midrule" + table_lines.extend( + [ + "\\toprule", + header_row, + "\\endfirsthead", + "\\toprule", + header_row, + "\\endhead", + "\\rowcolors{2}{gray!10}{white}", + ] + ) + + for index, risk in enumerate(risk_page.items): + severity = risk.attributes.get("severity", "") + probability = risk.attributes.get("probability", "") + residual_severity = risk.attributes.get("residual_severity", "") + residual_probability = risk.attributes.get("residual_probability", "") + label, score, colour = self._evaluate_risk_rating(severity, probability) + residual_label, residual_score, residual_colour = self._evaluate_risk_rating( + residual_severity, residual_probability + ) + risk_level_cell = self._format_risk_rating_cell(label, score, colour) + residual_level_cell = self._format_risk_rating_cell(residual_label, residual_score, residual_colour) + residual_risk_text = risk.attributes.get("residual_risk", "") + controls_text = self.process_text(risk.attributes.get("controls", "")) + + risk_expression = ( + f"{self.process_text(severity)} $\\times$ {self.process_text(probability)} = {risk_level_cell}" + ) + + control_lines = [controls_text] if controls_text else ["-"] + controls_content = " \\\\ ".join(control_lines) + controls_section = f"\\parbox[t]{{{controls_width}cm}}{{{controls_content}}}" + + residual_lines = [ + f"{self.process_text(residual_severity)} $\\times$ {self.process_text(residual_probability)}" + f" = {residual_level_cell if residual_level_cell else '-'}" + ] + if residual_risk_text: + residual_lines.append(self.process_text(residual_risk_text)) + residual_content = " \\\\ ".join([line for line in residual_lines if line] or ["-"]) + residual_cell = f"\\parbox[t]{{{residual_width}cm}}{{{residual_content}}}" + title_text = self.process_text(risk.title) + id_text = self.process_text_impl(risk.risk_id, set()) + first_cell = ( + f"\\phantomsection\\label{{{risk.risk_id}}}" + f"\\hyperref[{risk.risk_id}]{{{id_text}: {title_text}}}" + ) + row_cells = [ + first_cell, + self.process_text(risk.attributes.get("hazardous_situation", "")), + self.process_text(risk.attributes.get("harm", "")), + self.process_text(risk.attributes.get("cause", "")), + risk_expression, + controls_section, + residual_cell, + ] + row_ending = " \\\\ \\midrule" + if index == len(risk_page.items) - 1: + row_ending = " \\\\" + table_lines.append(" & ".join(row_cells) + row_ending) + + table_lines.extend( + [ + "\\bottomrule", + "\\end{longtable}", + "\\restoregeometry", + "\\setlength{\\paperwidth}{210mm}", + "\\setlength{\\paperheight}{297mm}", + "\\pdfpagewidth=210mm", + "\\pdfpageheight=297mm", + "\\special{papersize=210mm,297mm}", + "\\endgroup", + "\\clearpage", + ] + ) + return "\n".join(table_lines) + + def render_risk_document(self, risk_page: RiskDocument) -> str: + label = self._build_label_from_text("risk", risk_page.title) + latex = "\\section{" + self.process_text(risk_page.title) + "}\\label{" + label + "}\n\n" + latex += self.md_to_latex(risk_page.generic_content) + latex += "\n\n" + latex += self.build_risk_register(risk_page) + return latex + def md_to_latex(self, items: list[dict]) -> str: def handle_paragraph(item: dict) -> str: @@ -210,6 +434,12 @@ def handle_list(item: dict) -> str: latex.append("\n\\end{itemize}") return "".join(latex) + def handle_inline_html(item: dict) -> str: + text = item.get("text", "").strip().lower() + if text in {"
", "
", "
"}: + return "\\\\ " + return "" + def handle_image(item: dict) -> str: url = item["src"] latex = [ @@ -240,6 +470,37 @@ def handle_block_code(item: dict) -> str: return handle_playwright_test(item) return handle_code(item) + def render_children(children: list[dict]) -> str: + fragments: list[str] = [] + for child in children: + fragments.append(handle_item(child)) + return "".join(fragments) + + def handle_table(item: dict) -> str: + header_cells: list[str] = [] + body_rows: list[list[str]] = [] + for head_cell in item.get("children", [])[0]["children"]: + header_cells.append(render_children(head_cell.get("children", []))) + body_section = item.get("children", [])[1] + for row in body_section["children"]: + row_values = [render_children(cell.get("children", [])) for cell in row["children"]] + body_rows.append(row_values) + + column_spec = "|".join(["X"] * max(len(header_cells), 1)) + latex_lines = [ + "\\begin{table}[h]", + "\\centering", + "\\rowcolors{2}{gray!10}{white}", + f"\\begin{{tabularx}}{{\\linewidth}}{{|{column_spec}|}}", + "\\hline", + ] + latex_lines.append(" & ".join(f"\\textbf{{{cell}}}" for cell in header_cells) + " \\\\ \\hline") + for row in body_rows: + latex_lines.append(" & ".join(row) + " \\\\ \\hline") + latex_lines.append("\\end{tabularx}") + latex_lines.append("\\end{table}") + return "\n".join(latex_lines) + def handle_test_cover_page(_: dict) -> str: return r""" \begin{table}[h] @@ -381,9 +642,11 @@ def handle_item(item: dict) -> str: "heading": handle_heading, "list": handle_list, "image": handle_image, + "table": handle_table, "block_code": handle_block_code, "blank_line": lambda _: "\n", "newline": lambda _: "\n", + "inline_html": handle_inline_html, "strong": lambda item: f"\\textbf{{{self.process_text(item['children'][0]['text'])}}}", "emphasis": lambda item: f"\\emph{{{self.process_text(item['children'][0]['text'])}}}", "softbreak": lambda _: "\n", @@ -676,6 +939,9 @@ def __init__( for req_page in self.document.requirements: for requirement in req_page.items: self.unique_ids.add(requirement.req_id) + for risk_page in self.document.risks: + for risk in risk_page.items: + self.unique_ids.add(risk.risk_id) @staticmethod def _logo_basename(path: Optional[str], default_name: str) -> str: @@ -722,6 +988,15 @@ def render(self) -> bytes: with isolated_filesystem("report"): + for design_doc in self.document.design_documents: + document += self.render_markdown_document(design_doc) + + for supplementary_doc in self.document.supplementary_documents: + document += self.render_markdown_document(supplementary_doc) + + for risk_page in self.document.risks: + document += self.render_risk_document(risk_page) + for req_page in self.document.requirements: document += "\\section{" + req_page.title + "}\\label{" + req_page.title + "}\n\n" document += self.md_to_latex(req_page.generic_content) diff --git a/traceflow/res/report-header.tex b/traceflow/res/report-header.tex index a3b0e60..07eb5a6 100644 --- a/traceflow/res/report-header.tex +++ b/traceflow/res/report-header.tex @@ -1,12 +1,11 @@ \documentclass[10pt,a4paper]{article} \usepackage[T1]{fontenc} -\usepackage[paper=A4]{typearea} - \usepackage{geometry} \geometry{verbose,tmargin=3.0cm,bmargin=2.0cm,lmargin=1.5cm,rmargin=1.5cm,headheight=38pt} \renewcommand{\familydefault}{\sfdefault} \usepackage{tabularx} % in the preamble +\usepackage{longtable} \usepackage{array} \usepackage{hhline} \usepackage{amssymb}