workflow/tests/submit_endpoint_test.py

0001 """
0002 Offline check of the inline workflow-description submission endpoint.
0003
0004 Stubs the server-side dependencies (task buffer, workflow interface, request validation) so that
0005 pandaserver.api.v1.workflow_api.submit_workflow_description can be exercised without a database or
0006 a running server.
0007
0008 Run from the repository root:  python3 pandaserver/workflow/tests/submit_endpoint_test.py
0009 """
0010
0011 import copy
0012 import importlib.abc
0013 import importlib.machinery
0014 import json
0015 import os
0016 import sys
0017 import types
0018 from typing import Any, cast
0019
0020 REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
0021 sys.path.insert(0, REPO_ROOT)
0022
0023 # ---- stub what a bare checkout does not have -----------------------------------------------
0024 AUTO_STUB_ROOTS = ("idds", "pandaclient", "ruamel", "requests", "snakemake")
0025
0026
0027 class AutoStubFinder(importlib.abc.MetaPathFinder, importlib.abc.Loader):
0028     def find_spec(self, name, path=None, target=None):
0029         if name.split(".")[0] in AUTO_STUB_ROOTS:
0030             return importlib.machinery.ModuleSpec(name, self, is_package=True)
0031         return None
0032
0033     def create_module(self, spec):
0034         module = types.ModuleType(spec.name)
0035         module.__path__ = []
0036         return module
0037
0038     def exec_module(self, module):
0039         class Anything:
0040             def __init__(self, *args, **kwargs):
0041                 pass
0042
0043             def __call__(self, *args, **kwargs):
0044                 return self
0045
0046         module.__getattr__ = lambda name: Anything
0047
0048
0049 sys.meta_path.insert(0, AutoStubFinder())
0050
0051
0052 def stub(name, **attrs):
0053     module = types.ModuleType(name)
0054     for key, value in attrs.items():
0055         setattr(module, key, value)
0056     sys.modules[name] = module
0057     return module
0058
0059
0060 class QuietLogWrapper:
0061     def __init__(self, *args, **kwargs):
0062         self.messages = []
0063
0064     def info(self, message):
0065         pass
0066
0067     def debug(self, message):
0068         pass
0069
0070     def warning(self, message):
0071         self.messages.append(message)
0072
0073     def error(self, message):
0074         self.messages.append(message)
0075
0076
0077 import datetime
0078
0079 stub("pandacommon")
0080 pandautils = stub("pandacommon.pandautils")
0081 pandautils.__path__ = []
0082 stub("pandacommon.pandautils.base", SpecBase=object)
0083 stub(
0084     "pandacommon.pandautils.PandaUtils", naive_utcnow=lambda: datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None), get_sql_IN_bind_variables=None
0085 )
0086 pandalogger = stub("pandacommon.pandalogger")
0087 pandalogger.__path__ = []
0088 stub("pandacommon.pandalogger.LogWrapper", LogWrapper=QuietLogWrapper)
0089 stub("pandacommon.pandalogger.PandaLogger", PandaLogger=lambda: types.SimpleNamespace(getLogger=lambda name: None))
0090 stub("pandaserver.config", panda_config=types.SimpleNamespace(schemaJEDI="ATLAS_PANDA", schemaDEFT="ATLAS_DEFT"))
0091
0092 # the VOMS attributes reported to the endpoint; adjusted by the tests
0093 FAKE_ROLE = {"production": True}
0094 FAKE_FQANS = ["/atlas/Role=production", "/atlas/usatlas"]
0095
0096
0097 def fake_request_validation(logger, secure=False, production=False, request_method=None):
0098     def decorator(func):
0099         return func
0100
0101     return decorator
0102
0103
0104 stub(
0105     "pandaserver.api.v1.common",
0106     MESSAGE_DATABASE="database error",
0107     TIME_OUT="timeout",
0108     TimedMethod=object,
0109     generate_response=lambda success, message="", data=None: {"success": success, "message": message, "data": data},
0110     get_dn=lambda req: "/DC=ch/DC=cern/CN=test user",
0111     get_fqan=lambda req: list(FAKE_FQANS),
0112     has_production_role=lambda req: FAKE_ROLE["production"],
0113     request_validation=fake_request_validation,
0114 )
0115 stub("pandaserver.srvcore")
0116 stub("pandaserver.srvcore.panda_request", PandaRequest=object)
0117 stub("pandaserver.taskbuffer")
0118 stub("pandaserver.taskbuffer.TaskBuffer", TaskBuffer=object)
0119 stub("pandaserver.workflow.workflow_core", WorkflowInterface=lambda *a, **k: None)
0120
0121 from pandaserver.api.v1 import workflow_api  # noqa: E402
0122
0123
0124 # ---- fakes standing in for the task buffer and the workflow interface ----------------------
0125 class FakeTaskBuffer:
0126     def __init__(self, existing=None, fail=False):
0127         self.existing = existing or {}
0128         self.fail = fail
0129         self.queries = []
0130
0131     def get_existing_task_names(self, vo, prod_source_label, task_names):
0132         self.queries.append((vo, prod_source_label, sorted(task_names)))
0133         if self.fail:
0134             return None
0135         return {name: info for name, info in self.existing.items() if name in task_names}
0136
0137
0138 class FakeWorkflowInterface:
0139     def __init__(self, workflow_id=4242):
0140         self.workflow_id = workflow_id
0141         self.calls = []
0142         self.step_relations = {"workflow_id": 133, "steps": [{"step_id": 516, "name": "evgen", "parent_step_ids": []}]}
0143         self.task_relations = {"workflow_id": 133, "tasks": [{"key": "133:516", "task_id": 52382519, "name": "evgen", "parents": []}]}
0144         self.workflow_status = {
0145             "workflow_id": 133,
0146             "name": "mc23_526140_AZH_llbb_full_chain",
0147             "status": "failed",
0148             "reason": "step recon (step_id=525) failed: task 48810706 reported broken",
0149             "parent_id": None,
0150             "creation_time": "2026-10-12T09:14:03",
0151             "start_time": "2026-10-12T09:15:01",
0152             "end_time": "2026-10-12T11:02:44",
0153             "n_steps": 9,
0154             "step_status_counts": {"done": 4, "failed": 1, "cancelled": 4},
0155         }
0156
0157     def register_workflow(
0158         self, prodsourcelabel, user_dn, workflow_name=None, workflow_definition=None, raw_request_params=None, prod_role=False, fqans=None, *args, **kwargs
0159     ):
0160         self.calls.append(
0161             {
0162                 "prodsourcelabel": prodsourcelabel,
0163                 "user_dn": user_dn,
0164                 "workflow_name": workflow_name,
0165                 "raw_request_params": raw_request_params,
0166                 "prod_role": prod_role,
0167                 "fqans": fqans,
0168             }
0169         )
0170         return self.workflow_id
0171
0172     # The relation queries. Each returns whatever was put on the fake, so the endpoints can be
0173     # checked on how they pass arguments through and how they report nothing found.
0174     def get_step_relations(self, workflow_id):
0175         self.calls.append({"get_step_relations": workflow_id})
0176         return self.step_relations
0177
0178     def get_task_relations(self, workflow_id):
0179         self.calls.append({"get_task_relations": workflow_id})
0180         return self.task_relations
0181
0182     def get_task_relations_of_task(self, task_id):
0183         self.calls.append({"get_task_relations_of_task": task_id})
0184         return self.task_relations
0185
0186     def get_workflow_status(self, workflow_id):
0187         self.calls.append({"get_workflow_status": workflow_id})
0188         return self.workflow_status
0189
0190
0191 def install(existing=None, fail=False, workflow_id=4242):
0192     tbif = FakeTaskBuffer(existing, fail)
0193     wfif = FakeWorkflowInterface(workflow_id)
0194     # The module declares these as the real TaskBuffer and WorkflowInterface. The endpoints call
0195     # only the two methods the fakes carry, which is what the checks below are about.
0196     workflow_api.global_task_buffer = cast(Any, tbif)
0197     workflow_api.global_wfif = cast(Any, wfif)
0198     return tbif, wfif
0199
0200
0201 def check(label, condition, detail=""):
0202     print(f"  {'PASS' if condition else 'FAIL'}  {label}{'  ' + str(detail) if detail and not condition else ''}")
0203     return condition
0204
0205
0206 def main():
0207     failures = 0
0208     wfd = json.load(open(os.path.join(os.path.dirname(__file__), "..", "examples", "production_chain_wfd.json")))
0209     submit = workflow_api.submit_workflow_description
0210
0211     print("\n=== a valid description registers ===")
0212     tbif, wfif = install()
0213     res = submit(None, copy.deepcopy(wfd))
0214     failures += not check("success", res["success"] is True, res["message"])
0215     failures += not check("workflow_id returned", res["data"] == {"workflow_id": 4242}, res["data"])
0216     failures += not check("no warning message", res["message"] == "", res["message"])
0217     failures += not check("registered exactly once", len(wfif.calls) == 1)
0218     call = wfif.calls[0]
0219     failures += not check("workflow_name taken from the description", call["workflow_name"] == wfd["name"], call["workflow_name"])
0220     failures += not check("prodsourcelabel managed for a production role", call["prodsourcelabel"] == "managed", call["prodsourcelabel"])
0221     failures += not check(
0222         "description carried under the inline key",
0223         list(call["raw_request_params"]) == [workflow_api.INLINE_DESCRIPTION_KEY],
0224         list(call["raw_request_params"]),
0225     )
0226     failures += not check("no sandbox keys invented", "sandbox" not in call["raw_request_params"] and "sourceURL" not in call["raw_request_params"])
0227     failures += not check("production role captured from VOMS", call["prod_role"] is True, call["prod_role"])
0228     failures += not check("fqans captured from VOMS", call["fqans"] == FAKE_FQANS, call["fqans"])
0229
0230     print("\n=== credentials come from VOMS, never from the payload ===")
0231     spoofed = copy.deepcopy(wfd)
0232     spoofed["prod_role"] = True
0233     spoofed["fqans"] = ["/atlas/Role=production"]
0234     FAKE_ROLE["production"] = False
0235     FAKE_FQANS_SAVED = list(FAKE_FQANS)
0236     del FAKE_FQANS[:]
0237     FAKE_FQANS.extend(["/atlas"])
0238     tbif_s, wfif_s = install()
0239     submit(None, spoofed)
0240     failures += not check("payload cannot claim a production role", wfif_s.calls[0]["prod_role"] is False, wfif_s.calls[0]["prod_role"])
0241     failures += not check("payload cannot inject fqans", wfif_s.calls[0]["fqans"] == ["/atlas"], wfif_s.calls[0]["fqans"])
0242     failures += not check("payload cannot force prodsourcelabel", wfif_s.calls[0]["prodsourcelabel"] == "user")
0243     FAKE_ROLE["production"] = True
0244     del FAKE_FQANS[:]
0245     FAKE_FQANS.extend(FAKE_FQANS_SAVED)
0246
0247     print("\n=== the description is accepted as a JSON string too ===")
0248     tbif, wfif = install()
0249     res = submit(None, json.dumps(wfd))
0250     failures += not check("success", res["success"] is True, res["message"])
0251     res_bad = submit(None, "{not json")
0252     failures += not check("malformed JSON rejected", res_bad["success"] is False)
0253     failures += not check("malformed JSON explained", "Failed to parse" in res_bad["message"], res_bad["message"])
0254
0255     print("\n=== a non-production submitter gets prodsourcelabel user ===")
0256     FAKE_ROLE["production"] = False
0257     tbif, wfif = install()
0258     submit(None, copy.deepcopy(wfd))
0259     failures += not check("prodsourcelabel user", wfif.calls[0]["prodsourcelabel"] == "user", wfif.calls[0]["prodsourcelabel"])
0260     FAKE_ROLE["production"] = True
0261
0262     print("\n=== an invalid description is rejected before registration ===")
0263     broken = copy.deepcopy(wfd)
0264     for job_param in broken["steps"]["simul"]["task_params"]["jobParameters"]:
0265         if job_param.get("param_type") == "input":
0266             job_param["dataset"] = "{merge_evnt/NOPE}"
0267     tbif, wfif = install()
0268     res = submit(None, broken)
0269     failures += not check("rejected", res["success"] is False)
0270     failures += not check("reason reported", "does not match any step output" in res["message"], res["message"])
0271     failures += not check("nothing registered", wfif.calls == [])
0272     failures += not check("no duplication query wasted", tbif.queries == [])
0273
0274     print("\n=== a duplicated taskName warns but still registers ===")
0275     existing = {
0276         wfd["steps"]["evgen"]["task_params"]["taskName"]: {"jediTaskID": 48810693, "status": "running"},
0277         wfd["steps"]["merge_aod"]["task_params"]["taskName"]: {"jediTaskID": 48810733, "status": None},
0278     }
0279     tbif, wfif = install(existing=existing)
0280     res = submit(None, copy.deepcopy(wfd))
0281     failures += not check("still succeeds", res["success"] is True)
0282     failures += not check("still registers", len(wfif.calls) == 1)
0283     failures += not check("warning names both collisions", res["message"].count("jediTaskID=") == 2, res["message"])
0284     failures += not check("JEDI status reported", "status=running" in res["message"], res["message"])
0285     failures += not check("DEFT-only collision reported", "queued in DEFT" in res["message"], res["message"])
0286     failures += not check("queried once per vo/prodSourceLabel group", len(tbif.queries) == 1, tbif.queries)
0287     failures += not check("queried all 9 taskNames", len(tbif.queries[0][2]) == 9, tbif.queries[0][2])
0288     failures += not check("queried group scoped by vo and label", tbif.queries[0][:2] == ("atlas", "managed"), tbif.queries[0][:2])
0289
0290     print("\n=== a failed duplication lookup does not block submission ===")
0291     tbif, wfif = install(fail=True)
0292     res = submit(None, copy.deepcopy(wfd))
0293     failures += not check("still succeeds", res["success"] is True, res["message"])
0294     failures += not check("still registers", len(wfif.calls) == 1)
0295
0296     print("\n=== registration failure is reported ===")
0297     tbif, wfif = install(workflow_id=None)
0298     res = submit(None, copy.deepcopy(wfd))
0299     failures += not check("failure reported", res["success"] is False)
0300     failures += not check("reason given", "Failed to submit" in res["message"], res["message"])
0301
0302     print("\n=== the deprecated alias delegates to submit_workflow ===")
0303     raw_params = {"sourceURL": "https://example.org", "sandbox": "sandbox.tgz", "language": "yaml", "workflowSpecFile": "wf.yaml", "outDS": "user.me.out"}
0304     tbif, wfif = install()
0305     # request_validation is faked to a pass-through above, so no endpoint here ever reads the
0306     # request object; the alias is undecorated on purpose and so still declares its real type.
0307     res_alias = workflow_api.submit_workflow_raw_request(cast(Any, None), copy.deepcopy(raw_params))
0308     tbif2, wfif2 = install()
0309     res_direct = workflow_api.submit_workflow(None, copy.deepcopy(raw_params))
0310     failures += not check("alias succeeds", res_alias["success"] is True, res_alias["message"])
0311     failures += not check("alias result matches submit_workflow", res_alias == res_direct, (res_alias, res_direct))
0312     failures += not check("alias registers the raw request unchanged", wfif.calls[0]["raw_request_params"]["sandbox"] == "sandbox.tgz")
0313     failures += not check("raw request path does not validate a description", "workflow_description" not in wfif.calls[0]["raw_request_params"])
0314
0315     print("\n=== routing ===")
0316     print("\n=== get_step_relations ===")
0317     _, wfif = install()
0318     res = workflow_api.get_step_relations(cast(Any, None), 133)
0319     failures += not check("succeeds", res["success"] is True, res)
0320     failures += not check("the workflow id is passed through", wfif.calls[-1] == {"get_step_relations": 133}, wfif.calls[-1])
0321     failures += not check("the relations are the data", res["data"]["steps"][0]["name"] == "evgen", res["data"])
0322     _, wfif = install()
0323     wfif.step_relations = None
0324     res = workflow_api.get_step_relations(cast(Any, None), 999)
0325     failures += not check("nothing found is a failure with a reason", res["success"] is False and "No step" in res["message"], res)
0326
0327     print("\n=== get_workflow_status ===")
0328     _, wfif = install()
0329     res = workflow_api.get_workflow_status(cast(Any, None), 133)
0330     failures += not check("succeeds", res["success"] is True, res)
0331     failures += not check("the workflow id is passed through", wfif.calls[-1] == {"get_workflow_status": 133}, wfif.calls[-1])
0332     failures += not check("the workflow's own status is the data", res["data"]["status"] == "failed", res["data"])
0333     failures += not check("the step tally comes back with it", res["data"]["step_status_counts"]["cancelled"] == 4, res["data"])
0334     failures += not check("the reason names the step that failed", "recon" in res["data"]["reason"], res["data"]["reason"])
0335     failures += not check("the times are strings a plain json client can read", isinstance(res["data"]["end_time"], str), res["data"]["end_time"])
0336     _, wfif = install()
0337     wfif.workflow_status = None
0338     res = workflow_api.get_workflow_status(cast(Any, None), 999)
0339     failures += not check("an unknown workflow is a failure with a reason", res["success"] is False and "No workflow" in res["message"], res)
0340
0341     print("\n=== get_task_relations ===")
0342     _, wfif = install()
0343     res = workflow_api.get_task_relations(cast(Any, None), workflow_id=133)
0344     failures += not check("by workflow id", res["success"] is True and wfif.calls[-1] == {"get_task_relations": 133}, wfif.calls[-1])
0345     _, wfif = install()
0346     res = workflow_api.get_task_relations(cast(Any, None), task_id=52401216)
0347     failures += not check("by task id, through the other method", wfif.calls[-1] == {"get_task_relations_of_task": 52401216}, wfif.calls[-1])
0348     failures += not check("...and succeeds", res["success"] is True, res)
0349     _, wfif = install()
0350     res = workflow_api.get_task_relations(cast(Any, None))
0351     failures += not check("neither given is refused", res["success"] is False and "required" in res["message"], res)
0352     failures += not check("...without reaching the interface", not any("get_task" in str(call) for call in wfif.calls), wfif.calls)
0353     res = workflow_api.get_task_relations(cast(Any, None), workflow_id=133, task_id=52401216)
0354     failures += not check("both given is refused", res["success"] is False and "not both" in res["message"], res)
0355     _, wfif = install()
0356     wfif.task_relations = None
0357     res = workflow_api.get_task_relations(cast(Any, None), task_id=52401216)
0358     failures += not check("a task no workflow runs is a failure with a reason", res["success"] is False and "runs this task" in res["message"], res)
0359
0360     # extract_allowed_methods lives in the stubbed common module, so apply its rule directly:
0361     # module-level functions defined in this module whose name does not start with an underscore
0362     import inspect
0363
0364     exported = [
0365         name
0366         for name, obj in inspect.getmembers(workflow_api, inspect.isfunction)
0367         if obj.__module__ == workflow_api.__name__ and name != "init_task_buffer" and not name.startswith("_")
0368     ]
0369     print(f"    exported: {sorted(exported)}")
0370     failures_local = 0
0371     failures_local += not check("the new endpoint is routed", "submit_workflow_description" in exported)
0372     failures_local += not check(
0373         "all four endpoints routed",
0374         {"submit_workflow", "submit_workflow_description", "submit_workflow_definition", "submit_workflow_raw_request"} <= set(exported),
0375         sorted(exported),
0376     )
0377     failures_local += not check("the deprecated alias is still routed", "submit_workflow_raw_request" in exported)
0378     failures_local += not check(
0379         "the query endpoints are routed",
0380         {"get_step_relations", "get_task_relations", "get_workflow_status"} <= set(exported),
0381         sorted(exported),
0382     )
0383     failures_local += not check("private helpers not routed", not any(n.startswith("_") for n in exported))
0384     failures_local += not check(
0385         "imported helpers not routed",
0386         "validate_workflow_description" not in exported and "substitute_placeholder" not in exported,
0387     )
0388
0389     failures += failures_local
0390     print(f"\n{'ALL CHECKS PASSED' if not failures else f'{failures} CHECK(S) FAILED'}")
0391     return 1 if failures else 0
0392
0393
0394 if __name__ == "__main__":
0395     sys.exit(main())