workflow/examples/insert_step_task_test.py

0001 """
0002 Offline check of insert_step_task, the DEFT insert used for a workflow step's task.
0003
0004 Drives the real WorkflowModule.insert_step_task against a fake cursor, so the SQL it builds, the
0005 late-bound ${TASKID} resolution, the active-task throttle and both sequence backends are exercised
0006 without a database.
0007
0008 Run from the repository root:  python3 pandaserver/workflow/examples/insert_step_task_test.py
0009 """
0010
0011 import json
0012 import os
0013 import sys
0014 import types
0015 from typing import Any, cast
0016
0017 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")))
0018
0019
0020 def stub(n, **a):
0021     m = types.ModuleType(n)
0022     for k, v in a.items():
0023         setattr(m, k, v)
0024     sys.modules[n] = m
0025     return m
0026
0027
0028 stub("pandacommon")
0029 pu = stub("pandacommon.pandautils")
0030 pu.__path__ = []
0031 stub("pandacommon.pandautils.base", SpecBase=object)
0032
0033
0034 def get_sql_IN_bind_variables(vals, prefix=":", value_as_suffix=False):
0035     names = [f"{prefix}{v}" for v in vals]
0036     return ",".join(names), {n: v for n, v in zip(names, vals)}
0037
0038
0039 stub("pandacommon.pandautils.PandaUtils", get_sql_IN_bind_variables=get_sql_IN_bind_variables, naive_utcnow=lambda: None)
0040 pl = stub("pandacommon.pandalogger")
0041 pl.__path__ = []
0042
0043
0044 class Log:
0045     def __init__(s, *a, **k):
0046         s.msgs = []
0047
0048     def info(s, m):
0049         pass
0050
0051     def debug(s, m):
0052         s.msgs.append(m)
0053
0054     def warning(s, m):
0055         pass
0056
0057     def error(s, m):
0058         s.msgs.append("ERR:" + str(m))
0059
0060
0061 stub("pandacommon.pandalogger.LogWrapper", LogWrapper=Log)
0062 stub("pandacommon.pandalogger.PandaLogger", PandaLogger=lambda: types.SimpleNamespace(getLogger=lambda n: None))
0063 stub("pandaserver.config", panda_config=types.SimpleNamespace(schemaJEDI="ATLAS_PANDA", schemaDEFT="ATLAS_DEFT"))
0064 stub("pandaserver.srvcore")
0065 stub("pandaserver.srvcore.CoreUtils", clean_user_id=lambda d: "tester")
0066 stub("pandaserver.taskbuffer.ErrorCode")
0067 stub("pandaserver.taskbuffer.JobUtils")
0068 stub("pandaserver.taskbuffer.JobSpec", JobSpec=object)
0069
0070
0071 class BaseModule:
0072     _max: int | None = None
0073
0074     def __init__(self, log_stream=None):
0075         self.backend = "oracle"
0076
0077     def create_tagged_logger(self, comment, tag=""):
0078         return Log()
0079
0080     def getvalue_corrector(self, v):
0081         return v
0082
0083     def getConfigValue(self, comp, key, app=None, vo=None):
0084         return self._max
0085
0086     def dump_error_message(self, log):
0087         pass
0088
0089     def _commit(self):
0090         return True
0091
0092     def _rollback(self):
0093         pass
0094
0095
0096 stub("pandaserver.taskbuffer.db_proxy_mods.base_module", BaseModule=BaseModule, varNUMBER=int)
0097 stub("pandaserver.taskbuffer.db_proxy_mods.entity_module", get_entity_module=lambda s: None)
0098
0099 import contextlib
0100
0101 from pandaserver.taskbuffer.db_proxy_mods.workflow_module import WorkflowModule
0102 from pandaserver.workflow.workflow_base import TASKID_PLACEHOLDER
0103
0104
0105 class Cur:
0106     def __init__(s, task_id, active_count):
0107         s.task_id = task_id
0108         s.active_count = active_count
0109         s.executed = []
0110         s._last = None
0111
0112     def var(s, t):
0113         return "OUT"
0114
0115     def execute(s, sql, varmap=None):
0116         s.executed.append((sql, dict(varmap or {})))
0117         if "COUNT(*)" in sql:
0118             s._last = (s.active_count,)
0119
0120     def fetchone(s):
0121         return s._last
0122
0123     def getvalue(s, ref):
0124         return s.task_id
0125
0126
0127 class Proxy(WorkflowModule):
0128     def __init__(s, task_id=49900001, active_count=0, max_tasks=None):
0129         s.backend = "oracle"
0130         # The module under test reads self.cur, which the real BaseModule declares as a
0131         # WrappedCursor. fake_cur is the same object under a name the checks can read `executed`
0132         # from without claiming the stand-in is one.
0133         s.fake_cur = Cur(task_id, active_count)
0134         s.cur = cast(Any, s.fake_cur)
0135         s._max = max_tasks
0136
0137     def create_tagged_logger(s, comment, tag=""):
0138         return Log()
0139
0140     def getvalue_corrector(s, v):
0141         return v
0142
0143     def getConfigValue(s, comp, key, app=None, vo=None):
0144         return s._max
0145
0146     def dump_error_message(s, log):
0147         pass
0148
0149     def _commit(s):
0150         return True
0151
0152     def _rollback(s, useOtherError=False):
0153         return True
0154
0155     @contextlib.contextmanager
0156     def transaction(s, name=None, tmp_log=None):
0157         yield (s.cur, tmp_log or Log())
0158
0159
0160 FAILURES: list[str] = []
0161
0162
0163 def ok(label, condition, detail=""):
0164     if not condition:
0165         FAILURES.append(label)
0166     print(f"  {'PASS' if condition else 'FAIL'}  {label}{'  ' + str(detail) if detail and not condition else ''}")
0167
0168
0169 base = {
0170     "taskName": "mc23.evgen.e8590",
0171     "vo": "atlas",
0172     "prodSourceLabel": "managed",
0173     "userName": "mnegrini",
0174     "taskPriority": 275,
0175     "jobParameters": [
0176         {
0177             "type": "template",
0178             "param_type": "output",
0179             "dataset": f"mc23.evgen.EVNT.e8590_wfid12345_tid{TASKID_PLACEHOLDER}_00",
0180             "value": f"--outputEVNTFile=EVNT.{TASKID_PLACEHOLDER}._${{SN}}.pool.root",
0181         }
0182     ],
0183 }
0184
0185 print("insert_step_task:")
0186 p = Proxy()
0187 tid, msg = p.insert_step_task(dict(base), "/DC=ch/CN=t")
0188 ok("returns the new task id", tid == 49900001, (tid, msg))
0189 sqls = [e[0] for e in p.fake_cur.executed]
0190 ok("one INSERT into T_TASK", sum("INSERT INTO ATLAS_DEFT.T_TASK" in q for q in sqls) == 1, sqls)
0191 ok("followed by the placeholder UPDATE", sum("UPDATE ATLAS_DEFT.T_TASK SET jedi_task_parameters" in q for q in sqls) == 1)
0192 ins = [e for e in p.fake_cur.executed if "INSERT INTO ATLAS_DEFT.T_TASK" in e[0]][0]
0193 ok("uses the sequence for taskid", "PRODSYS2_TASK_ID_SEQ.nextval" in ins[0])
0194 ok("own id as parent when no parent given", "PRODSYS2_TASK_ID_SEQ.currval" in ins[0])
0195 ok("priority carried from taskPriority", ins[1][":priority"] == 275 and ins[1][":current_priority"] == 275)
0196 ok("userName taken as authored, not the DN", ins[1][":userName"] == "mnegrini")
0197 ok("status queued as waiting", ins[1][":status"] == "waiting")
0198 ok("inserted params still hold the placeholder", TASKID_PLACEHOLDER in ins[1][":param"])
0199 upd = [e for e in p.fake_cur.executed if "UPDATE" in e[0]][0]
0200 ok("update resolves it to the real id", TASKID_PLACEHOLDER not in upd[1][":param"] and "tid49900001_00" in upd[1][":param"])
0201 ok("update also resolves it in the LFN template", "EVNT.49900001._${SN}" in upd[1][":param"])
0202 ok("update targets the new task", upd[1][":taskid"] == 49900001)
0203
0204 print("\nno placeholder -> no extra UPDATE:")
0205 clean = json.loads(json.dumps(base).replace(TASKID_PLACEHOLDER, "48810693"))
0206 p2 = Proxy()
0207 tid2, _ = p2.insert_step_task(clean, "/DC=ch/CN=t")
0208 ok("still inserts", tid2 == 49900001)
0209 ok("no UPDATE issued", not any("UPDATE" in q for q, _ in p2.fake_cur.executed))
0210
0211 print("\nexplicit parent_tid:")
0212 p3 = Proxy()
0213 p3.insert_step_task(dict(base), "/DC=ch/CN=t", parent_tid=48810693)
0214 ins3 = [e for e in p3.fake_cur.executed if "INSERT INTO" in e[0]][0]
0215 ok("binds the given parent", ins3[1][":parent_tid"] == 48810693 and "currval" not in ins3[0])
0216
0217 print("\nactive task throttle:")
0218 p4 = Proxy(active_count=500, max_tasks=100)
0219 tid4, msg4 = p4.insert_step_task(dict(base), "/DC=ch/CN=t")
0220 ok("refused when over the limit", tid4 is None)
0221 ok("message explains it", "too many active tasks" in msg4, msg4)
0222 ok("nothing inserted", not any("INSERT INTO ATLAS_DEFT.T_TASK" in q for q, _ in p4.fake_cur.executed))
0223 p5 = Proxy(active_count=50, max_tasks=100)
0224 tid5, _ = p5.insert_step_task(dict(base), "/DC=ch/CN=t")
0225 ok("allowed when under the limit", tid5 == 49900001)
0226 p6 = Proxy(active_count=999, max_tasks=None)
0227 tid6, _ = p6.insert_step_task(dict(base), "/DC=ch/CN=t")
0228 ok("no limit configured -> no throttle query", tid6 == 49900001 and not any("COUNT(*)" in q for q, _ in p6.fake_cur.executed))
0229
0230 print("\nmissing required params:")
0231 for key in ["taskName", "vo", "prodSourceLabel", "userName"]:
0232     bad = dict(base)
0233     del bad[key]
0234     t, m = Proxy().insert_step_task(bad, "/DC=ch/CN=t")
0235     ok(f"rejects missing {key}", t is None and key in m, m)
0236
0237 print("\nmysql-style backend (no sequence):")
0238
0239
0240 class MyProxy(Proxy):
0241     def __init__(s, **k):
0242         super().__init__(**k)
0243         s.backend = "mysql"
0244
0245
0246 class MyCur(Cur):
0247     def execute(s, sql, varmap=None):
0248         s.executed.append((sql, dict(varmap or {})))
0249         if "COUNT(*)" in sql:
0250             s._last = (s.active_count,)
0251         elif "LAST_INSERT_ID" in sql:
0252             s._last = (49900001,)
0253
0254
0255 p7 = MyProxy()
0256 p7.fake_cur = MyCur(49900001, 0)
0257 p7.cur = cast(Any, p7.fake_cur)
0258 tid7, msg7 = p7.insert_step_task(dict(base), "/DC=ch/CN=t")
0259 ok("inserts via the stand-in sequence table", tid7 == 49900001, msg7)
0260 q7 = [e for e in p7.fake_cur.executed if "INSERT INTO ATLAS_DEFT.T_TASK" in e[0]][0]
0261 ok("binds nextval instead of a sequence", q7[1].get(":nextval") == 49900001 and "PRODSYS2_TASK_ID_SEQ.nextval" not in q7[0])
0262 ok("own id as parent uses the same value", ":nextval" in q7[0].split("VALUES")[1] and "currval" not in q7[0])
0263
0264 print(f"\n{'ALL CHECKS PASSED' if not FAILURES else f'{len(FAILURES)} CHECK(S) FAILED'}")
0265 sys.exit(1 if FAILURES else 0)