workflow/examples/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/examples/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
0145     def register_workflow(
0146         self, prodsourcelabel, user_dn, workflow_name=None, workflow_definition=None, raw_request_params=None, prod_role=False, fqans=None, *args, **kwargs
0147     ):
0148         self.calls.append(
0149             {
0150                 "prodsourcelabel": prodsourcelabel,
0151                 "user_dn": user_dn,
0152                 "workflow_name": workflow_name,
0153                 "raw_request_params": raw_request_params,
0154                 "prod_role": prod_role,
0155                 "fqans": fqans,
0156             }
0157         )
0158         return self.workflow_id
0159
0160     # The relation queries. Each returns whatever was put on the fake, so the endpoints can be
0161     # checked on how they pass arguments through and how they report nothing found.
0162     def get_step_relations(self, workflow_id):
0163         self.calls.append({"get_step_relations": workflow_id})
0164         return self.step_relations
0165
0166     def get_task_relations(self, workflow_id):
0167         self.calls.append({"get_task_relations": workflow_id})
0168         return self.task_relations
0169
0170     def get_task_relations_of_task(self, task_id):
0171         self.calls.append({"get_task_relations_of_task": task_id})
0172         return self.task_relations
0173
0174
0175 def install(existing=None, fail=False, workflow_id=4242):
0176     tbif = FakeTaskBuffer(existing, fail)
0177     wfif = FakeWorkflowInterface(workflow_id)
0178     # The module declares these as the real TaskBuffer and WorkflowInterface. The endpoints call
0179     # only the two methods the fakes carry, which is what the checks below are about.
0180     workflow_api.global_task_buffer = cast(Any, tbif)
0181     workflow_api.global_wfif = cast(Any, wfif)
0182     return tbif, wfif
0183
0184
0185 def check(label, condition, detail=""):
0186     print(f"  {'PASS' if condition else 'FAIL'}  {label}{'  ' + str(detail) if detail and not condition else ''}")
0187     return condition
0188
0189
0190 def main():
0191     failures = 0
0192     wfd = json.load(open(os.path.join(os.path.dirname(__file__), "production_chain_wfd.json")))
0193     submit = workflow_api.submit_workflow_description
0194
0195     print("\n=== a valid description registers ===")
0196     tbif, wfif = install()
0197     res = submit(None, copy.deepcopy(wfd))
0198     failures += not check("success", res["success"] is True, res["message"])
0199     failures += not check("workflow_id returned", res["data"] == {"workflow_id": 4242}, res["data"])
0200     failures += not check("no warning message", res["message"] == "", res["message"])
0201     failures += not check("registered exactly once", len(wfif.calls) == 1)
0202     call = wfif.calls[0]
0203     failures += not check("workflow_name taken from the description", call["workflow_name"] == wfd["name"], call["workflow_name"])
0204     failures += not check("prodsourcelabel managed for a production role", call["prodsourcelabel"] == "managed", call["prodsourcelabel"])
0205     failures += not check(
0206         "description carried under the inline key",
0207         list(call["raw_request_params"]) == [workflow_api.INLINE_DESCRIPTION_KEY],
0208         list(call["raw_request_params"]),
0209     )
0210     failures += not check("no sandbox keys invented", "sandbox" not in call["raw_request_params"] and "sourceURL" not in call["raw_request_params"])
0211     failures += not check("production role captured from VOMS", call["prod_role"] is True, call["prod_role"])
0212     failures += not check("fqans captured from VOMS", call["fqans"] == FAKE_FQANS, call["fqans"])
0213
0214     print("\n=== credentials come from VOMS, never from the payload ===")
0215     spoofed = copy.deepcopy(wfd)
0216     spoofed["prod_role"] = True
0217     spoofed["fqans"] = ["/atlas/Role=production"]
0218     FAKE_ROLE["production"] = False
0219     FAKE_FQANS_SAVED = list(FAKE_FQANS)
0220     del FAKE_FQANS[:]
0221     FAKE_FQANS.extend(["/atlas"])
0222     tbif_s, wfif_s = install()
0223     submit(None, spoofed)
0224     failures += not check("payload cannot claim a production role", wfif_s.calls[0]["prod_role"] is False, wfif_s.calls[0]["prod_role"])
0225     failures += not check("payload cannot inject fqans", wfif_s.calls[0]["fqans"] == ["/atlas"], wfif_s.calls[0]["fqans"])
0226     failures += not check("payload cannot force prodsourcelabel", wfif_s.calls[0]["prodsourcelabel"] == "user")
0227     FAKE_ROLE["production"] = True
0228     del FAKE_FQANS[:]
0229     FAKE_FQANS.extend(FAKE_FQANS_SAVED)
0230
0231     print("\n=== the description is accepted as a JSON string too ===")
0232     tbif, wfif = install()
0233     res = submit(None, json.dumps(wfd))
0234     failures += not check("success", res["success"] is True, res["message"])
0235     res_bad = submit(None, "{not json")
0236     failures += not check("malformed JSON rejected", res_bad["success"] is False)
0237     failures += not check("malformed JSON explained", "Failed to parse" in res_bad["message"], res_bad["message"])
0238
0239     print("\n=== a non-production submitter gets prodsourcelabel user ===")
0240     FAKE_ROLE["production"] = False
0241     tbif, wfif = install()
0242     submit(None, copy.deepcopy(wfd))
0243     failures += not check("prodsourcelabel user", wfif.calls[0]["prodsourcelabel"] == "user", wfif.calls[0]["prodsourcelabel"])
0244     FAKE_ROLE["production"] = True
0245
0246     print("\n=== an invalid description is rejected before registration ===")
0247     broken = copy.deepcopy(wfd)
0248     for job_param in broken["steps"]["simul"]["task_params"]["jobParameters"]:
0249         if job_param.get("param_type") == "input":
0250             job_param["dataset"] = "{merge_evnt/NOPE}"
0251     tbif, wfif = install()
0252     res = submit(None, broken)
0253     failures += not check("rejected", res["success"] is False)
0254     failures += not check("reason reported", "does not match any step output" in res["message"], res["message"])
0255     failures += not check("nothing registered", wfif.calls == [])
0256     failures += not check("no duplication query wasted", tbif.queries == [])
0257
0258     print("\n=== a duplicated taskName warns but still registers ===")
0259     existing = {
0260         wfd["steps"]["evgen"]["task_params"]["taskName"]: {"jediTaskID": 48810693, "status": "running"},
0261         wfd["steps"]["merge_aod"]["task_params"]["taskName"]: {"jediTaskID": 48810733, "status": None},
0262     }
0263     tbif, wfif = install(existing=existing)
0264     res = submit(None, copy.deepcopy(wfd))
0265     failures += not check("still succeeds", res["success"] is True)
0266     failures += not check("still registers", len(wfif.calls) == 1)
0267     failures += not check("warning names both collisions", res["message"].count("jediTaskID=") == 2, res["message"])
0268     failures += not check("JEDI status reported", "status=running" in res["message"], res["message"])
0269     failures += not check("DEFT-only collision reported", "queued in DEFT" in res["message"], res["message"])
0270     failures += not check("queried once per vo/prodSourceLabel group", len(tbif.queries) == 1, tbif.queries)
0271     failures += not check("queried all 9 taskNames", len(tbif.queries[0][2]) == 9, tbif.queries[0][2])
0272     failures += not check("queried group scoped by vo and label", tbif.queries[0][:2] == ("atlas", "managed"), tbif.queries[0][:2])
0273
0274     print("\n=== a failed duplication lookup does not block submission ===")
0275     tbif, wfif = install(fail=True)
0276     res = submit(None, copy.deepcopy(wfd))
0277     failures += not check("still succeeds", res["success"] is True, res["message"])
0278     failures += not check("still registers", len(wfif.calls) == 1)
0279
0280     print("\n=== registration failure is reported ===")
0281     tbif, wfif = install(workflow_id=None)
0282     res = submit(None, copy.deepcopy(wfd))
0283     failures += not check("failure reported", res["success"] is False)
0284     failures += not check("reason given", "Failed to submit" in res["message"], res["message"])
0285
0286     print("\n=== the deprecated alias delegates to submit_workflow ===")
0287     raw_params = {"sourceURL": "https://example.org", "sandbox": "sandbox.tgz", "language": "yaml", "workflowSpecFile": "wf.yaml", "outDS": "user.me.out"}
0288     tbif, wfif = install()
0289     # request_validation is faked to a pass-through above, so no endpoint here ever reads the
0290     # request object; the alias is undecorated on purpose and so still declares its real type.
0291     res_alias = workflow_api.submit_workflow_raw_request(cast(Any, None), copy.deepcopy(raw_params))
0292     tbif2, wfif2 = install()
0293     res_direct = workflow_api.submit_workflow(None, copy.deepcopy(raw_params))
0294     failures += not check("alias succeeds", res_alias["success"] is True, res_alias["message"])
0295     failures += not check("alias result matches submit_workflow", res_alias == res_direct, (res_alias, res_direct))
0296     failures += not check("alias registers the raw request unchanged", wfif.calls[0]["raw_request_params"]["sandbox"] == "sandbox.tgz")
0297     failures += not check("raw request path does not validate a description", "workflow_description" not in wfif.calls[0]["raw_request_params"])
0298
0299     print("\n=== routing ===")
0300     print("\n=== get_step_relations ===")
0301     _, wfif = install()
0302     res = workflow_api.get_step_relations(cast(Any, None), 133)
0303     failures += not check("succeeds", res["success"] is True, res)
0304     failures += not check("the workflow id is passed through", wfif.calls[-1] == {"get_step_relations": 133}, wfif.calls[-1])
0305     failures += not check("the relations are the data", res["data"]["steps"][0]["name"] == "evgen", res["data"])
0306     _, wfif = install()
0307     wfif.step_relations = None
0308     res = workflow_api.get_step_relations(cast(Any, None), 999)
0309     failures += not check("nothing found is a failure with a reason", res["success"] is False and "No step" in res["message"], res)
0310
0311     print("\n=== get_task_relations ===")
0312     _, wfif = install()
0313     res = workflow_api.get_task_relations(cast(Any, None), workflow_id=133)
0314     failures += not check("by workflow id", res["success"] is True and wfif.calls[-1] == {"get_task_relations": 133}, wfif.calls[-1])
0315     _, wfif = install()
0316     res = workflow_api.get_task_relations(cast(Any, None), task_id=52401216)
0317     failures += not check("by task id, through the other method", wfif.calls[-1] == {"get_task_relations_of_task": 52401216}, wfif.calls[-1])
0318     failures += not check("...and succeeds", res["success"] is True, res)
0319     _, wfif = install()
0320     res = workflow_api.get_task_relations(cast(Any, None))
0321     failures += not check("neither given is refused", res["success"] is False and "required" in res["message"], res)
0322     failures += not check("...without reaching the interface", not any("get_task" in str(call) for call in wfif.calls), wfif.calls)
0323     res = workflow_api.get_task_relations(cast(Any, None), workflow_id=133, task_id=52401216)
0324     failures += not check("both given is refused", res["success"] is False and "not both" in res["message"], res)
0325     _, wfif = install()
0326     wfif.task_relations = None
0327     res = workflow_api.get_task_relations(cast(Any, None), task_id=52401216)
0328     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)
0329
0330     # extract_allowed_methods lives in the stubbed common module, so apply its rule directly:
0331     # module-level functions defined in this module whose name does not start with an underscore
0332     import inspect
0333
0334     exported = [
0335         name
0336         for name, obj in inspect.getmembers(workflow_api, inspect.isfunction)
0337         if obj.__module__ == workflow_api.__name__ and name != "init_task_buffer" and not name.startswith("_")
0338     ]
0339     print(f"    exported: {sorted(exported)}")
0340     failures_local = 0
0341     failures_local += not check("the new endpoint is routed", "submit_workflow_description" in exported)
0342     failures_local += not check(
0343         "all four endpoints routed",
0344         {"submit_workflow", "submit_workflow_description", "submit_workflow_definition", "submit_workflow_raw_request"} <= set(exported),
0345         sorted(exported),
0346     )
0347     failures_local += not check("the deprecated alias is still routed", "submit_workflow_raw_request" in exported)
0348     failures_local += not check(
0349         "the relation endpoints are routed",
0350         {"get_step_relations", "get_task_relations"} <= set(exported),
0351         sorted(exported),
0352     )
0353     failures_local += not check("private helpers not routed", not any(n.startswith("_") for n in exported))
0354     failures_local += not check(
0355         "imported helpers not routed",
0356         "validate_workflow_description" not in exported and "substitute_placeholder" not in exported,
0357     )
0358
0359     failures += failures_local
0360     print(f"\n{'ALL CHECKS PASSED' if not failures else f'{failures} CHECK(S) FAILED'}")
0361     return 1 if failures else 0
0362
0363
0364 if __name__ == "__main__":
0365     sys.exit(main())