Coverage for tests/dag_test_utils.py: 98%
57 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 01:13 +0000
1# This file is part of ctrl_bps_htcondor.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
28"""Utility functions for ctrl_bps_htcondor tests."""
30import logging
31import re
32from pathlib import Path
34from htcondor2 import Schedd, classad
36from lsst.ctrl.bps.htcondor.lssthtc import HTC_VALID_JOB_KEYS, HTCDag, HTCJob, RestrictedDict
38_LOG = logging.getLogger(__name__)
41LOCATE_SUCCESS = classad.ClassAd("""[
42 CondorPlatform = "$CondorPlatform: X86_64-CentOS_7.9 $";
43 MyType = "Scheduler";
44 Machine = "testmachine";
45 Name = "testmachine";
46 CondorVersion = "$CondorVersion: 23.0.3 2024-04-04 $";
47 MyAddress = "<127.0.0.1:9618?addrs=127.0.0.1-9618+snip>"
48 ]
49""")
51PING_SUCCESS = classad.ClassAd("""[
52 AuthCommand = 60011;
53 AuthMethods = "FS_REMOTE";
54 Command = 60040;
55 AuthorizationSucceeded = true;
56 ValidCommands = "60002,60003,60011,60014,60045,60046,60047,60048,60049,60050,60052,523";
57 TriedAuthentication = true;
58 RemoteVersion = "$CondorVersion: 10.9.0 2023-09-28 BuildID: 678228 PackageID: 10.9.0-1 $";
59 MyRemoteUserName = "testuser@testmachine";
60 Authentication = "YES";
61 ]
62""")
65class ScheddMatch:
66 """Matches any Schedd object whose address equals the expected
67 address string. Useful for assert_called_once_with.
69 Parameters
70 ----------
71 schedd_classad : `str`
72 Expected value to be compared against in __eq__.
73 """
75 def __init__(self, schedd_classad: classad.ClassAd):
76 self.addr_str = schedd_classad["MyAddress"]
78 def __eq__(self, other_schedd: Schedd):
79 return other_schedd._addr == self.addr_str
82class ConstraintMatch:
83 """Matches any constraint string against the expected pattern
84 (regular expressions allowed). Useful for assert_called_once_with.
86 Parameters
87 ----------
88 expected : `str`
89 Expected value to be compared against in __eq__.
90 """
92 def __init__(self, expected: str):
93 self.expected_pattern = expected
95 def __eq__(self, other: object):
96 if not isinstance(other, str): # pragma: no cover
97 return NotImplemented
98 return re.match(self.expected_pattern, other) is not None
101def make_lazy_dag(name: str, final: bool = True) -> tuple[HTCDag, list[str]]:
102 """Manually make a simple HTCDag with a lazy subdag for unit testing.
104 Parameters
105 ----------
106 name : `str`
107 Name to use for HTCDag.
108 final : `bool`, optional
109 Whether to include a FINAL job. Defaults to True.
111 Returns
112 -------
113 dag : `lsst.ctrl.bps.htcondor.lssthtc.HTCDag`
114 A simple HTCDag with a lazy subdag for testing.
115 filelist : `list` [`str`]
116 A list of files with relative paths that should be created
117 when DAG is written.
118 """
119 dag = HTCDag(name=name)
121 job_cmds = RestrictedDict(HTC_VALID_JOB_KEYS, {"executable": "/usr/bin/uptime"})
122 job_attrs = {"bps_run": "test_run"}
124 job = HTCJob("job1", "label1", job_cmds, initattrs=job_attrs)
125 subdir = Path(f"jobs/{job.label}")
126 job.subdir = subdir
127 job.subfile = f"{job.name}.sub"
128 job.add_dag_cmds({"dir": subdir})
129 dag.add_node("job1", data=job)
131 job = HTCJob("make_subdag", "label2", job_cmds, initattrs=job_attrs)
132 subdir = Path(f"jobs/{job.label}")
133 job.subdir = subdir
134 job.subfile = f"{job.name}.sub"
135 job.add_dag_cmds({"dir": subdir})
136 dag.add_node(job.name, data=job)
137 dag.add_edge("job1", job.name)
139 job = HTCJob("lazy_subdag", "lazy_label", job_cmds, initattrs=job_attrs)
140 job.subfile = "lazy_subdag.dag.condor.sub"
141 job.dag = HTCDag(name="lazy_subdag")
142 dag.add_node(job.name, data=job)
143 dag.add_edge("make_subdag", job.name)
145 job = HTCJob("job4", "label4", job_cmds, initattrs=job_attrs)
146 subdir = Path(f"jobs/{job.label}")
147 job.subdir = subdir
148 job.subfile = f"{job.name}.sub"
149 job.add_dag_cmds({"dir": subdir})
150 dag.add_node(job.name, data=job)
151 dag.add_edge("lazy_subdag", job.name)
153 filelist = [
154 f"{name}.dag",
155 "lazy_subdag.dag.condor.sub",
156 "jobs/label1/job1.sub",
157 "jobs/label2/make_subdag.sub",
158 "jobs/label4/job4.sub",
159 ]
161 if final: 161 ↛ 170line 161 didn't jump to line 170 because the condition on line 161 was always true
162 job = HTCJob("finalJob", "finalJob", job_cmds, initattrs=job_attrs)
163 subdir = Path(f"jobs/{job.label}")
164 job.subdir = subdir
165 job.subfile = f"{job.name}.sub"
166 job.add_dag_cmds({"dir": subdir})
167 dag.graph["final_job"] = job
168 filelist.append("jobs/finalJob/finalJob.sub")
170 return dag, filelist