diff --git a/src/soa_builder/web/app.py b/src/soa_builder/web/app.py index 9c473a7..ed8318c 100644 --- a/src/soa_builder/web/app.py +++ b/src/soa_builder/web/app.py @@ -56,6 +56,7 @@ _migrate_element_table, _migrate_rename_cell_table, _migrate_rollback_add_elements_restored, + _migrate_add_epoch_type, ) from .routers import activities as activities_router from .routers import arms as arms_router @@ -67,7 +68,7 @@ from .routers.arms import create_arm # re-export for backward compatibility from .routers.arms import delete_arm from .schemas import ArmCreate, SOACreate, SOAMetadataUpdate -from .utils import get_next_code_uid as _get_next_code_uid +from .utils import get_next_code_uid as _get_next_code_uid, load_epoch_type_options load_dotenv() # must come BEFORE reading env-based configuration so values are populated DB_PATH = os.environ.get("SOA_BUILDER_DB", "soa_builder_web.db") @@ -105,6 +106,7 @@ # Database migration steps +_migrate_add_epoch_type() _migrate_add_arm_uid() _migrate_drop_arm_element_link() _migrate_add_epoch_id_to_visit() @@ -1697,7 +1699,9 @@ def ui_refresh_concepts(request: Request, soa_id: int): if request.headers.get("HX-Request") == "true": return HTMLResponse("", headers={"HX-Redirect": f"/ui/soa/{soa_id}/edit"}) # Fallback: plain form POST non-htmx redirect via script - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) """Freeze & rollback endpoints moved to routers/freezes.py and routers/rollback.py""" @@ -2777,7 +2781,24 @@ def delete_epoch(soa_id: int, epoch_id: int): "epoch_label": b[4], "epoch_description": b[5], } - cur.execute("DELETE FROM epoch WHERE id=", (epoch_id,)) + # Include current type in before snapshot + try: + cur.execute("SELECT type FROM epoch WHERE id=?", (epoch_id,)) + tr = cur.fetchone() + if before is not None: + before["type"] = tr[0] if tr else None + except Exception: + pass + # Clear visit epoch references to avoid dangling links + try: + cur.execute( + "UPDATE visit SET epoch_id=NULL WHERE soa_id=? AND epoch_id=?", + (soa_id, epoch_id), + ) + except Exception: + pass + # Delete the epoch row + cur.execute("DELETE FROM epoch WHERE id=?", (epoch_id,)) conn.commit() conn.close() _reindex("epoch", soa_id) @@ -2848,7 +2869,9 @@ def ui_add_activity(request: Request, soa_id: int, name: str = Form(...)): "activity_uid": f"Activity_{order_index}", }, ) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/create", response_class=HTMLResponse) @@ -2934,7 +2957,9 @@ def ui_update_meta( ) conn.commit() conn.close() - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.get("/ui/soa/{soa_id}/edit", response_class=HTMLResponse) @@ -3132,11 +3157,31 @@ def ui_edit(request: Request, soa_id: int): } ) + # Admin audit view: recent activity audits for this SOA + conn_activity_audit = _connect() + cur_activity_audit = conn_activity_audit.cursor() + cur_activity_audit.execute( + "SELECT id, activity_id, action, before_json, after_json, performed_at FROM activity_audit WHERE soa_id=? ORDER BY id DESC LIMIT 20", + (soa_id,), + ) + activity_audits = [ + { + "id": r[0], + "activity_id": r[1], + "action": r[2], + "before_json": r[3], + "after_json": r[4], + "performed_at": r[5], + } + for r in cur_activity_audit.fetchall() + ] + conn_activity_audit.close() + # Admin audit view: recent arm audits for this SOA - conn_audit = _connect() - cur_audit = conn_audit.cursor() - cur_audit.execute( - "SELECT id, arm_id, action, before_json, after_json, performed_at FROM arm_audit WHERE soa_id=? ORDER BY id DESC LIMIT 50", + conn_arm_audit = _connect() + cur_arm_audit = conn_arm_audit.cursor() + cur_arm_audit.execute( + "SELECT id, arm_id, action, before_json, after_json, performed_at FROM arm_audit WHERE soa_id=? ORDER BY id DESC LIMIT 20", (soa_id,), ) arm_audits = [ @@ -3148,9 +3193,78 @@ def ui_edit(request: Request, soa_id: int): "after_json": r[4], "performed_at": r[5], } - for r in cur_audit.fetchall() + for r in cur_arm_audit.fetchall() + ] + conn_arm_audit.close() + + # Admin audit view: recent epoch audits for this SoA + conn_epoch_audit = _connect() + cur_epoch_audit = conn_epoch_audit.cursor() + cur_epoch_audit.execute( + "SELECT id, epoch_id, action, before_json, after_json, performed_at FROM epoch_audit WHERE soa_id=? ORDER BY id DESC LIMIT 20", + (soa_id,), + ) + epoch_audits = [ + { + "id": r[0], + "epoch_id": r[1], + "action": r[2], + "before_json": r[3], + "after_json": r[4], + "performed_at": r[5], + } + for r in cur_epoch_audit.fetchall() + ] + conn_epoch_audit.close() + + # Enrich epochs using API-only map: code -> submissionValue + # Resolve stored epoch.type (code_uid) to terminology code via code table, then map to submissionValue. + code_map: dict[int, str] = {} + conn_em = _connect() + cur_em = conn_em.cursor() + cur_em.execute( + "SELECT e.id, c.code FROM epoch e LEFT JOIN code c ON c.code_uid = e.type AND c.soa_id = e.soa_id WHERE e.soa_id=?", + (soa_id,), + ) + for eid, code in cur_em.fetchall(): + if eid is not None and code: + code_map[eid] = code + conn_em.close() + try: + from .utils import load_epoch_type_map + + code_to_submission = load_epoch_type_map(force=False) or {} + except Exception: + code_to_submission = {} + epochs = [ + { + **e, + "epoch_type_submission_value": code_to_submission.get( + code_map.get(e.get("id"), ""), None + ), + } + for e in epochs ] - conn_audit.close() + + # Epoch Type options (C99079) must come from CDISC API only + epoch_type_options = load_epoch_type_options(force=False) or [] + logger.info( + "Epoch Type options (API only) count=%d values=%s", + len(epoch_type_options), + ", ".join(epoch_type_options) if epoch_type_options else "", + ) + # Additional diagnostics + try: + from . import utils as _u + + logger.info( + "Epoch Type diagnostics last_status=%s last_url=%s last_error=%s", + _u._epoch_type_cache.get("last_status"), + _u._epoch_type_cache.get("last_url"), + _u._epoch_type_cache.get("last_error"), + ) + except Exception: + pass return templates.TemplateResponse( request, @@ -3176,6 +3290,10 @@ def ui_edit(request: Request, soa_id: int): "protocol_terminology_C174222": protocol_terminology_C174222, "ddf_terminology_C188727": ddf_terminology_C188727, "arm_audits": arm_audits, + "epoch_audits": epoch_audits, + "activity_audits": activity_audits, + # Epoch Type options (C99079) + "epoch_type_options": epoch_type_options, }, ) @@ -3509,7 +3627,9 @@ def ui_add_visit( "epoch_id": parsed_epoch, }, ) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/add_arm", response_class=HTMLResponse) @@ -3632,7 +3752,7 @@ async def ui_add_arm( # Properly escape the value for safety in HTML/JS context escaped_selection = json.dumps(data_origin_type_submission) return HTMLResponse( - f"", + f"", status_code=400, ) # Create Code_N (continue numbering) @@ -3759,7 +3879,7 @@ async def ui_update_arm( ) conn.close() return HTMLResponse( - f"", + f"", status_code=400, ) @@ -3819,7 +3939,7 @@ async def ui_update_arm( ) conn.close() return HTMLResponse( - f"", + f"", status_code=400, ) # Maintain/Upsert immutable Code_N for DDF mapping @@ -3935,13 +4055,17 @@ async def ui_update_arm( arm_id, ) conn.close() - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/delete_arm", response_class=HTMLResponse) def ui_delete_arm(request: Request, soa_id: int, arm_id: int = Form(...)): delete_arm(soa_id, arm_id) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/reorder_arms", response_class=HTMLResponse) @@ -4067,7 +4191,9 @@ def ui_add_element( "element_id": element_identifier, }, ) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/update_element", response_class=HTMLResponse) @@ -4148,7 +4274,9 @@ def ui_update_element( before=before, after={**after, "updated_fields": updated_fields}, ) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/delete_element", response_class=HTMLResponse) @@ -4164,7 +4292,9 @@ def ui_delete_element(request: Request, soa_id: int, element_id: int = Form(...) _record_element_audit( soa_id, "delete", element_id, before={"id": element_id}, after=None ) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/add_epoch", response_class=HTMLResponse) @@ -4174,6 +4304,7 @@ def ui_add_epoch( name: str = Form(...), epoch_label: Optional[str] = Form(None), epoch_description: Optional[str] = Form(None), + epoch_type_submission_value: Optional[str] = Form(None), ): """Form handler to add an Epoch.""" if not _soa_exists(soa_id): @@ -4185,8 +4316,42 @@ def ui_add_epoch( cur.execute("SELECT MAX(epoch_seq) FROM epoch WHERE soa_id=?", (soa_id,)) row = cur.fetchone() next_seq = (row[0] or 0) + 1 + # Optional epoch type mapping via code junction (C99079) using API-only map + epoch_type_submission_value = (epoch_type_submission_value or "").strip() or None + selected_code_uid = None + if epoch_type_submission_value: + try: + from .utils import load_epoch_type_map, get_epoch_parent_package_href_cached + + epoch_map = load_epoch_type_map() + except Exception: + epoch_map = {} + # Invert map to find conceptId by submissionValue + concept_id = None + for cid, sv in (epoch_map or {}).items(): + if sv and sv.strip().lower() == epoch_type_submission_value.strip().lower(): + concept_id = cid + break + if concept_id: + # Create a new Code_N for this conceptId under C99079 (API-only) + code_uid = _get_next_code_uid(cur, soa_id) + try: + parent_href = get_epoch_parent_package_href_cached() or None + except Exception: + parent_href = None + cur.execute( + "INSERT INTO code (soa_id, code_uid, codelist_table, codelist_code, code) VALUES (?,?,?,?,?)", + ( + soa_id, + code_uid, + parent_href, + "C99079", + concept_id, + ), + ) + selected_code_uid = code_uid cur.execute( - "INSERT INTO epoch (soa_id,name,order_index,epoch_seq,epoch_label,epoch_description) VALUES (?,?,?,?,?,?)", + "INSERT INTO epoch (soa_id,name,order_index,epoch_seq,epoch_label,epoch_description,type) VALUES (?,?,?,?,?,?,?)", ( soa_id, name, @@ -4194,6 +4359,7 @@ def ui_add_epoch( next_seq, (epoch_label or "").strip() or None, (epoch_description or "").strip() or None, + selected_code_uid, ), ) eid = cur.lastrowid @@ -4203,7 +4369,7 @@ def ui_add_epoch( soa_id, "create", eid, - before=None, + before={"type": None}, after={ "id": eid, "name": name, @@ -4211,9 +4377,12 @@ def ui_add_epoch( "epoch_seq": next_seq, "epoch_label": (epoch_label or "").strip() or None, "epoch_description": (epoch_description or "").strip() or None, + "type": selected_code_uid, }, ) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/update_epoch", response_class=HTMLResponse) @@ -4224,6 +4393,7 @@ def ui_update_epoch( name: Optional[str] = Form(None), epoch_label: Optional[str] = Form(None), epoch_description: Optional[str] = Form(None), + epoch_type_submission_value: Optional[str] = Form(None), ): """Form handler to update an existing Epoch.""" if not _soa_exists(soa_id): @@ -4254,6 +4424,17 @@ def ui_update_epoch( "epoch_label": b[4], "epoch_description": b[5], } + # Include current type in before snapshot for audit + try: + conn_bt = _connect() + cur_bt = conn_bt.cursor() + cur_bt.execute("SELECT type FROM epoch WHERE id=?", (epoch_id,)) + br = cur_bt.fetchone() + conn_bt.close() + if before is not None: + before["type"] = br[0] if br else None + except Exception: + pass sets = [] vals: list[Any] = [] if name is not None: @@ -4265,6 +4446,59 @@ def ui_update_epoch( if epoch_description is not None: sets.append("epoch_description=?") vals.append((epoch_description or "").strip() or None) + # Handle epoch type mapping via code junction (C99079) using API-only map + epoch_type_submission_value = (epoch_type_submission_value or "").strip() or None + if epoch_type_submission_value is not None: + # If empty string provided, clear type + if epoch_type_submission_value == "": + sets.append("type=?") + vals.append(None) + else: + # Resolve submission value to conceptId via API-only map + try: + from .utils import ( + load_epoch_type_map, + get_epoch_parent_package_href_cached, + ) + + epoch_map = load_epoch_type_map() + except Exception: + epoch_map = {} + concept_id = None + for cid, sv in (epoch_map or {}).items(): + if ( + sv + and sv.strip().lower() + == epoch_type_submission_value.strip().lower() + ): + concept_id = cid + break + selected_code_uid = None + if concept_id: + conn_t = _connect() + cur_t = conn_t.cursor() + # Always create a new Code_N for C99079 selections (no reuse) + code_uid = _get_next_code_uid(cur_t, soa_id) + try: + parent_href = get_epoch_parent_package_href_cached() or None + except Exception: + parent_href = None + cur_t.execute( + "INSERT INTO code (soa_id, code_uid, codelist_table, codelist_code, code) VALUES (?,?,?,?,?)", + ( + soa_id, + code_uid, + parent_href, + "C99079", + concept_id, + ), + ) + selected_code_uid = code_uid + conn_t.commit() + conn_t.close() + # Persist epoch.type even if concept_id not found will be None + sets.append("type=?") + vals.append(selected_code_uid) if sets: conn_u = _connect() cur_u = conn_u.cursor() @@ -4287,7 +4521,16 @@ def ui_update_epoch( "epoch_seq": r[3], "epoch_label": r[4], "epoch_description": r[5], + "type": None, } + # Fetch type from epoch for audit after snapshot + conn_ta = _connect() + cur_ta = conn_ta.cursor() + cur_ta.execute("SELECT type FROM epoch WHERE id=?", (epoch_id,)) + tr_after = cur_ta.fetchone() + conn_ta.close() + if tr_after: + after_api["type"] = tr_after[0] _record_epoch_audit( soa_id, "update", @@ -4295,7 +4538,9 @@ def ui_update_epoch( before=before, after=after_api, ) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post( @@ -4331,7 +4576,9 @@ def ui_set_activity_concepts( edit=False, ) return HTMLResponse(html) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.get( @@ -4446,7 +4693,9 @@ def ui_delete_visit(request: Request, soa_id: int, visit_id: int = Form(...)): logger.error( "ui_delete_visit failed visit_id=%s soa_id=%s error=%s", visit_id, soa_id, e ) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/set_visit_epoch", response_class=HTMLResponse) @@ -4492,21 +4741,27 @@ def ui_set_visit_epoch( DB_PATH, ) conn.close() - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/delete_activity", response_class=HTMLResponse) def ui_delete_activity(request: Request, soa_id: int, activity_id: int = Form(...)): """Form handler to delete an Activity""" delete_activity(soa_id, activity_id) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/delete_epoch", response_class=HTMLResponse) def ui_delete_epoch(request: Request, soa_id: int, epoch_id: int = Form(...)): """Form handler to delete an Epoch.""" delete_epoch(soa_id, epoch_id) - return HTMLResponse(f"") + return HTMLResponse( + f"" + ) @app.post("/ui/soa/{soa_id}/reorder_visits", response_class=HTMLResponse) @@ -4587,12 +4842,27 @@ def ui_reorder_epochs(request: Request, soa_id: int, order: str = Form("")): conn.commit() conn.close() _record_reorder_audit(soa_id, "epoch", old_order, ids) + # Also record epoch-specific reorder audit for parity with JSON endpoint + def _epoch_types_snapshot(soa_id_int: int) -> list[dict]: + conn_s = _connect() + cur_s = conn_s.cursor() + cur_s.execute( + "SELECT id,type FROM epoch WHERE soa_id=? ORDER BY order_index", + (soa_id_int,), + ) + rows = cur_s.fetchall() + conn_s.close() + return [{"id": rid, "type": rtype} for rid, rtype in rows] + _record_epoch_audit( soa_id, "reorder", epoch_id=None, - before={"old_order": old_order}, + before={ + "old_order": old_order, + "types": _epoch_types_snapshot(soa_id), + }, after={"new_order": ids}, ) return HTMLResponse("OK") diff --git a/src/soa_builder/web/migrate_database.py b/src/soa_builder/web/migrate_database.py index 444364d..16fc06e 100644 --- a/src/soa_builder/web/migrate_database.py +++ b/src/soa_builder/web/migrate_database.py @@ -255,6 +255,30 @@ def _migrate_add_epoch_label_desc(): logger.warning("Epoch label/description migration failed: %s", e) +# Migrate: add epoch type (options from SDTM CT codelist_code=C99079) +def _migrate_add_epoch_type(): + """Add optional epoch type column if missing""" + try: + conn = _connect() + cur = conn.cursor() + cur.execute("PRAGMA table_info(epoch)") + cols = {r[1] for r in cur.fetchall()} + alters = [] + if "type" not in cols: + alters.append("ALTER TABLE epoch ADD COLUMN type TEXT") + for statement in alters: + try: + cur.execute(statement) + except Exception as e: + logger.warning("Failed epoch type migration '%s': %s", statement, e) + if alters: + conn.commit() + logger.info("Applied epoch type migration: %s", ", ".join(alters)) + conn.close() + except Exception as e: + logger.warning("Epoch type migration failed: %s", e) + + # Migration: create code_junction table def _migrate_create_code_junction(): """Create code_junction linking table if absent. diff --git a/src/soa_builder/web/routers/epochs.py b/src/soa_builder/web/routers/epochs.py index e33110f..3e00146 100644 --- a/src/soa_builder/web/routers/epochs.py +++ b/src/soa_builder/web/routers/epochs.py @@ -79,12 +79,13 @@ def add_epoch(soa_id: int, payload: EpochCreate): eid = cur.lastrowid conn.commit() conn.close() - result = {"epoch_id": eid, "order_index": order_index, "epoch_seq": next_seq} + + # Correct audit for create (type not set via JSON API) _record_epoch_audit( soa_id, "create", eid, - before=None, + before={"type": None}, after={ "id": eid, "name": payload.name, @@ -94,7 +95,6 @@ def add_epoch(soa_id: int, payload: EpochCreate): "epoch_description": (payload.epoch_description or "").strip() or None, }, ) - return result @router.get("/soa/{soa_id}/epochs") @@ -172,6 +172,14 @@ def update_epoch_metadata(soa_id: int, epoch_id: int, payload: EpochUpdate): "epoch_label": b[4], "epoch_description": b[5], } + # Include current type in before snapshot + try: + cur.execute("SELECT type FROM epoch WHERE id=?", (epoch_id,)) + tr = cur.fetchone() + if before is not None: + before["type"] = tr[0] if tr else None + except Exception: + pass sets = [] vals = [] if payload.name is not None: @@ -232,11 +240,26 @@ def reorder_epochs_api(soa_id: int, order: List[int]): cur.execute("UPDATE epoch SET order_index=? WHERE id=?", (idx, eid)) conn.commit() conn.close() + + def _epoch_types_snapshot_router(soa_id_int: int) -> List[dict]: + conn_s = _connect() + cur_s = conn_s.cursor() + cur_s.execute( + "SELECT id,type FROM epoch WHERE soa_id=? ORDER BY order_index", + (soa_id_int,), + ) + rows = cur_s.fetchall() + conn_s.close() + return [{"id": rid, "type": rtype} for rid, rtype in rows] + _record_epoch_audit( soa_id, "reorder", epoch_id=None, - before={"old_order": old_order}, + before={ + "old_order": old_order, + "types": _epoch_types_snapshot_router(soa_id), + }, after={"new_order": order}, ) return JSONResponse({"ok": True, "old_order": old_order, "new_order": order}) diff --git a/src/soa_builder/web/templates/edit.html b/src/soa_builder/web/templates/edit.html index 31ec5eb..2e6e134 100644 --- a/src/soa_builder/web/templates/edit.html +++ b/src/soa_builder/web/templates/edit.html @@ -152,6 +152,12 @@

Editing SoA {{ soa_id }}

+ @@ -161,6 +167,12 @@

Editing SoA {{ soa_id }}

+ @@ -265,9 +277,35 @@

Editing SoA {{ soa_id }}

-
+
+ Activity Audit (latest {{ activity_audits|length }}) + {% if activity_audits %} + + + + + + + + + + {% for au in activity_audits %} + + + + + + + + + {% endfor %} +
IDActivityActionPerformedBeforeAfter
{{ au.id }}{{ au.activity_id }}{{ au.action }}{{ au.performed_at }}{{ au.before_json or '' }}{{ au.after_json or '' }}
+ {% else %} +
No activity audit entries yet.
+ {% endif %} +
Arm Audit (latest {{ arm_audits|length }}) {% if arm_audits %} @@ -295,6 +333,33 @@

Editing SoA {{ soa_id }}

No arm audit entries yet.
{% endif %}
+
+ Epoch Audit (latest {{ epoch_audits|length }}) + {% if epoch_audits %} + + + + + + + + + + {% for au in epoch_audits %} + + + + + + + + + {% endfor %} +
IDEpochActionPerformedBeforeAfter
{{ au.id }}{{ au.epoch_id }}{{ au.action }}{{ au.performed_at }}{{ au.before_json or '' }}{{ au.after_json or '' }}
+ {% else %} +
No epoch audit entries yet.
+ {% endif %} +
diff --git a/src/soa_builder/web/utils.py b/src/soa_builder/web/utils.py index 87e37f7..78049d6 100644 --- a/src/soa_builder/web/utils.py +++ b/src/soa_builder/web/utils.py @@ -1,4 +1,213 @@ -from typing import Any +from typing import Any, Dict, List +import os +import requests +import time + +_epoch_type_cache: dict[str, Any] = { + "data": None, + "fetched_at": 0, + "last_status": None, + "last_url": None, + "last_error": None, + "parent_package_href": None, +} +_EPOCH_TYPE_CACHE_TTL = 60 * 60 # 1 hour + + +def load_epoch_type_options(force: bool = False) -> list[str]: + """Fetch Epoch Type options from CDISC Library API codelist C99079. + + Parses _links.terms[].submissionValue and returns a sorted, deduplicated list. + Uses env `CDISC_SUBSCRIPTION_KEY` and `_get_cdisc_api_key`-style headers when available. + Note: This module does not import app helpers; callers should provide headers if overriding. + """ + now = time.time() + if ( + not force + and _epoch_type_cache["data"] + and now - _epoch_type_cache["fetched_at"] < _EPOCH_TYPE_CACHE_TTL + ): + return _epoch_type_cache["data"] or [] + # Use only the specified CDISC Library endpoint (per user requirement) + url = "https://library.cdisc.org/api/mdr/ct/packages/sdtmct-2025-09-26/codelists/C99079" + headers: dict[str, str] = {"Accept": "application/json"} + subscription_key = os.environ.get("CDISC_SUBSCRIPTION_KEY") + api_key = os.environ.get("CDISC_API_KEY") or os.environ.get( + "CDISC_SUBSCRIPTION_KEY" + ) + unified_key = subscription_key or api_key + if unified_key: + headers["Ocp-Apim-Subscription-Key"] = unified_key + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + headers["api-key"] = api_key + try: + values: list[str] = [] + last_status = None + _epoch_type_cache.update(last_url=url, last_error=None) + resp = requests.get(url, headers=headers, timeout=10) + last_status = resp.status_code + if resp.status_code != 200: + data = {} + top_terms = [] + else: + data = resp.json() or {} + # Preferred structure: top-level 'terms' list + top_terms = [] + if isinstance(data, dict) and isinstance(data.get("terms"), list): + top_terms = data.get("terms") or [] + elif isinstance(data, list): + top_terms = data + else: + # HAL-style fallbacks + embedded_terms = [] + if isinstance(data.get("_embedded"), dict): + embedded_terms = data.get("_embedded", {}).get("terms", []) or [] + link_terms = data.get("_links", {}).get("terms", []) or [] + top_terms = embedded_terms or link_terms + # Capture parent package href if present + try: + if isinstance(data, dict): + pph = data.get("_links", {}).get("parentPackage", {}).get("href") + if pph: + _epoch_type_cache["parent_package_href"] = str(pph) + except Exception: + pass + # Collect embedded submissionValue + for t in top_terms: + if not isinstance(t, dict): + continue + sv = t.get("submissionValue") or t.get("cdisc_submission_value") + if sv and str(sv).strip(): + values.append(str(sv).strip()) + # If still none and we have term links, follow them + if not values: + for t in top_terms: + href = None + if isinstance(t, dict): + href = t.get("href") or t.get("_href") + if not href: + continue + try: + _epoch_type_cache.update(last_url=href) + term_resp = requests.get(href, headers=headers, timeout=10) + if term_resp.status_code == 200: + term_json = term_resp.json() or {} + sv = term_json.get("submissionValue") or term_json.get( + "cdisc_submission_value" + ) + if sv and str(sv).strip(): + values.append(str(sv).strip()) + except Exception: + continue + # Single endpoint only; no loop/break + result = sorted(list(dict.fromkeys(values))) + _epoch_type_cache.update(data=result, fetched_at=now, last_status=last_status) + return result + except Exception as e: + _epoch_type_cache.update(data=[], fetched_at=now, last_error=str(e)) + return [] + + +def load_epoch_type_map(force: bool = False) -> Dict[str, str]: + """Fetch Epoch Type term mapping from CDISC Library API for C99079. + + Returns a dict of {term_code: submissionValue}. This enables UI preselection + by mapping stored epoch.type code_uid -> code -> submissionValue. + """ + now = time.time() + # Simple TTL cache to avoid repeated remote calls + if not force and isinstance(_epoch_type_cache.get("_map"), dict): + cached_map = _epoch_type_cache.get("_map") or {} + fetched = _epoch_type_cache.get("_map_fetched_at") or 0 + if cached_map and now - fetched < _EPOCH_TYPE_CACHE_TTL: + return cached_map + + # Use only the specified CDISC Library endpoint (per user requirement) + headers: dict[str, str] = {"Accept": "application/json"} + subscription_key = os.environ.get("CDISC_SUBSCRIPTION_KEY") + api_key = os.environ.get("CDISC_API_KEY") or os.environ.get( + "CDISC_SUBSCRIPTION_KEY" + ) + unified_key = subscription_key or api_key + if unified_key: + headers["Ocp-Apim-Subscription-Key"] = unified_key + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + headers["api-key"] = api_key + + url = "https://library.cdisc.org/api/mdr/ct/packages/sdtmct-2025-09-26/codelists/C99079" + code_to_submission: Dict[str, str] = {} + last_status = None + try: + _epoch_type_cache.update(last_url=url, last_error=None) + resp = requests.get(url, headers=headers, timeout=10) + last_status = resp.status_code + if resp.status_code != 200: + data = {} + terms = [] + else: + data = resp.json() or {} + # Preferred structure: top-level 'terms' list + terms: List[dict] = [] + if isinstance(data, dict) and isinstance(data.get("terms"), list): + terms = data.get("terms") or [] + elif isinstance(data, list): + terms = data + else: + # HAL-style fallback + terms = data.get("_links", {}).get("terms", []) or [] + # Capture parent package href if present + try: + if isinstance(data, dict): + pph = data.get("_links", {}).get("parentPackage", {}).get("href") + if pph: + _epoch_type_cache["parent_package_href"] = str(pph) + except Exception: + pass + for t in terms: + if not isinstance(t, dict): + continue + # CDISC Library returns conceptId + submissionValue in this package endpoint + code = t.get("conceptId") or t.get("code") or t.get("termCode") + sub = t.get("submissionValue") or t.get("cdisc_submission_value") + if code and sub: + code_to_submission[str(code)] = str(sub).strip() + continue + href = t.get("href") or t.get("_href") + if not href: + # HAL style: try _links.self.href + linkself = t.get("_links", {}).get("self", {}) + href = linkself.get("href") if isinstance(linkself, dict) else None + if href: + try: + _epoch_type_cache.update(last_url=href) + term_resp = requests.get(href, headers=headers, timeout=10) + if term_resp.status_code == 200: + tj = term_resp.json() or {} + sub2 = tj.get("submissionValue") or tj.get( + "cdisc_submission_value" + ) + code2 = tj.get("code") or code + if code2 and sub2: + code_to_submission[str(code2)] = str(sub2).strip() + except Exception: + pass + # Single endpoint only; no loop/break + except Exception as e: + _epoch_type_cache.update(last_error=str(e)) + _epoch_type_cache.update(last_status=last_status) + _epoch_type_cache.update(_map=code_to_submission, _map_fetched_at=now) + return code_to_submission + + +def get_epoch_parent_package_href_cached() -> str | None: + """Return cached parentPackage href from the last Epoch Type API fetch. + + This depends on a prior call to load_epoch_type_options/map to populate the cache. + """ + val = _epoch_type_cache.get("parent_package_href") + return str(val) if val else None def get_next_code_uid(cur: Any, soa_id: int) -> str: diff --git a/tests/test_epoch_reorder_audit_api.py b/tests/test_epoch_reorder_audit_api.py new file mode 100644 index 0000000..2af639c --- /dev/null +++ b/tests/test_epoch_reorder_audit_api.py @@ -0,0 +1,85 @@ +import os +import json +import sqlite3 +from typing import List + +from fastapi.testclient import TestClient + +from soa_builder.web.app import app + + +client = TestClient(app) + + +def _db_path() -> str: + return os.environ.get("SOA_BUILDER_DB", "soa_builder_web.db") + + +def _fetch_epoch_audits(soa_id: int) -> List[dict]: + conn = sqlite3.connect(_db_path()) + cur = conn.cursor() + cur.execute( + "SELECT action, before_json, after_json FROM epoch_audit WHERE soa_id=? ORDER BY id", + (soa_id,), + ) + rows = cur.fetchall() + conn.close() + return [ + { + "action": r[0], + "before": json.loads(r[1]) if r[1] else None, + "after": json.loads(r[2]) if r[2] else None, + } + for r in rows + ] + + +def _create_soa(name="EpochReorderAuditAPI") -> int: + r = client.post("/soa", json={"name": name}) + assert r.status_code == 200 + return r.json()["id"] + + +def test_epoch_reorder_audit_api_structure(): + soa_id = _create_soa() + # Create three epochs via JSON API router + for idx, nm in enumerate(["E1", "E2", "E3"], start=1): + r_add = client.post( + f"/soa/{soa_id}/epochs", + json={ + "name": nm, + "epoch_label": f"L{idx}", + "epoch_description": f"D{idx}", + }, + ) + assert r_add.status_code == 200 + + # List epochs to get current order + r_list = client.get(f"/soa/{soa_id}/epochs") + assert r_list.status_code == 200 + epochs = r_list.json()["epochs"] + assert len(epochs) == 3 + old_order = [e["id"] for e in epochs] + + # New order: reverse + new_order = list(reversed(old_order)) + r_reorder = client.post(f"/soa/{soa_id}/epochs/reorder", json=new_order) + assert r_reorder.status_code == 200 + + audits = _fetch_epoch_audits(soa_id) + # Find the last reorder audit + reorder_audits = [a for a in audits if a["action"] == "reorder"] + assert len(reorder_audits) >= 1 + last = reorder_audits[-1] + assert last["before"] is not None and last["after"] is not None + # Validate before.old_order and before.types exist and types is a list of {id,type} + assert "old_order" in last["before"] + assert last["before"]["old_order"] == old_order + assert "types" in last["before"] + assert isinstance(last["before"]["types"], list) + if last["before"]["types"]: + sample = last["before"]["types"][0] + assert set(sample.keys()) == {"id", "type"} + # Validate after.new_order equals our new_order + assert "new_order" in last["after"] + assert last["after"]["new_order"] == new_order diff --git a/tests/test_epoch_type_audit.py b/tests/test_epoch_type_audit.py new file mode 100644 index 0000000..187b4b9 --- /dev/null +++ b/tests/test_epoch_type_audit.py @@ -0,0 +1,118 @@ +import os +import json +import sqlite3 +from pathlib import Path +from unittest.mock import patch + +from fastapi.testclient import TestClient + +from soa_builder.web.app import app + + +client = TestClient(app) + + +def _db_path(): + return os.environ.get("SOA_BUILDER_DB", str(Path("soa_builder_web.db").absolute())) + + +def _fetch_epoch_audits(soa_id: int): + conn = sqlite3.connect(_db_path()) + cur = conn.cursor() + cur.execute( + "SELECT action, before_json, after_json FROM epoch_audit WHERE soa_id=? ORDER BY id", + (soa_id,), + ) + rows = cur.fetchall() + conn.close() + return [ + { + "action": r[0], + "before": json.loads(r[1]) if r[1] else None, + "after": json.loads(r[2]) if r[2] else None, + } + for r in rows + ] + + +def _create_soa(name="AuditTest"): + r = client.post("/soa", json={"name": name}) + assert r.status_code == 200 + return r.json()["id"] + + +def test_epoch_type_audit_create_update_delete_contains_type(): + soa_id = _create_soa() + + # Mock the epoch type map to resolve submission values to conceptIds + fake_map = {"CID1": "TREATMENT", "CID2": "FOLLOW-UP"} + # Also mock parent package href to avoid relying on network + with ( + patch("soa_builder.web.utils.load_epoch_type_map", return_value=fake_map), + patch( + "soa_builder.web.utils.get_epoch_parent_package_href_cached", + return_value="https://library.cdisc.org/api/mdr/ct/packages/sdtmct-2025-09-26", + ), + ): + # Create epoch with type 'TREATMENT' + r_add = client.post( + f"/ui/soa/{soa_id}/add_epoch", + data={ + "name": "Screening", + "epoch_label": "SCR", + "epoch_description": "Initial screening", + "epoch_type_submission_value": "TREATMENT", + }, + ) + assert r_add.status_code == 200 + + audits = _fetch_epoch_audits(soa_id) + assert any(a["action"] == "create" for a in audits) + create_audit = next(a for a in audits if a["action"] == "create") + assert create_audit["after"] is not None + # Type should be present in after snapshot (code_uid value) + assert "type" in create_audit["after"] + created_type_uid = create_audit["after"]["type"] + assert created_type_uid is None or str(created_type_uid).startswith("Code_") + + # Update epoch type to 'FOLLOW-UP' and ensure audit after contains type + # Need the epoch_id for update; read from list epochs + rl = client.get(f"/soa/{soa_id}/epochs") + assert rl.status_code == 200 + epoch_id = rl.json()["epochs"][0]["id"] + + r_up = client.post( + f"/ui/soa/{soa_id}/update_epoch", + data={ + "epoch_id": epoch_id, + "name": "Screening", + "epoch_label": "SCR", + "epoch_description": "Initial screening", + "epoch_type_submission_value": "FOLLOW-UP", + }, + ) + assert r_up.status_code == 200 + + audits2 = _fetch_epoch_audits(soa_id) + assert any(a["action"] == "update" for a in audits2) + update_audit = [a for a in audits2 if a["action"] == "update"][-1] + assert update_audit["after"] is not None + assert "type" in update_audit["after"] + updated_type_uid = update_audit["after"]["type"] + # Should allow same or new; our handler creates new Code_N, so expect change + if created_type_uid: + assert updated_type_uid != created_type_uid + + # Delete epoch and ensure delete audit before has type + r_del = client.post( + f"/ui/soa/{soa_id}/delete_epoch", + data={"epoch_id": epoch_id}, + ) + assert r_del.status_code == 200 + + audits3 = _fetch_epoch_audits(soa_id) + assert any(a["action"] == "delete" for a in audits3) + delete_audit = [a for a in audits3 if a["action"] == "delete"][-1] + assert delete_audit["before"] is not None + assert "type" in delete_audit["before"] + assert delete_audit["before"]["type"] is not None diff --git a/tests/test_epoch_type_options.py b/tests/test_epoch_type_options.py new file mode 100644 index 0000000..295577b --- /dev/null +++ b/tests/test_epoch_type_options.py @@ -0,0 +1,57 @@ +import os +from unittest.mock import patch, Mock + +from soa_builder.web.utils import load_epoch_type_options + + +def test_load_epoch_type_options_parses_submission_values(): + # Ensure environment headers won't block test when absent + os.environ.pop("CDISC_SUBSCRIPTION_KEY", None) + os.environ.pop("CDISC_API_KEY", None) + + fake_json = { + "_links": { + "terms": [ + {"submissionValue": "SCREENING"}, + {"submissionValue": "TREATMENT"}, + {"submissionValue": "FOLLOW-UP"}, + {"submissionValue": "TREATMENT"}, # duplicate to test dedupe + ] + } + } + + mock_resp = Mock() + mock_resp.status_code = 200 + mock_resp.json.return_value = fake_json + + with patch("soa_builder.web.utils.requests.get", return_value=mock_resp) as pg: + values = load_epoch_type_options(force=True) + # Verify API called + assert pg.called + # Values should be deduplicated and sorted + assert values == ["FOLLOW-UP", "SCREENING", "TREATMENT"] + + +def test_load_epoch_type_options_caches_results(): + os.environ.pop("CDISC_SUBSCRIPTION_KEY", None) + os.environ.pop("CDISC_API_KEY", None) + + fake_json = {"_links": {"terms": [{"submissionValue": "A"}]}} + mock_resp = Mock(status_code=200) + mock_resp.json.return_value = fake_json + + with patch("soa_builder.web.utils.requests.get", return_value=mock_resp) as pg: + first = load_epoch_type_options(force=True) + second = load_epoch_type_options(force=False) + # Only one API call due to cache on the second call + assert pg.call_count == 1 + assert first == ["A"] + assert second == ["A"] + + +def test_load_epoch_type_options_handles_error_status_code(): + mock_resp = Mock(status_code=500) + mock_resp.json.return_value = {} + with patch("soa_builder.web.utils.requests.get", return_value=mock_resp): + values = load_epoch_type_options(force=True) + assert values == []