File indexing completed on 2025-01-30 09:14:56
0001 import sys
0002 import os
0003 import re
0004 from pathlib import Path
0005
0006 import pytest
0007
0008 import acts
0009 import acts.examples
0010
0011 pytestmark = [
0012 pytest.mark.skipif(
0013 sys.platform != "linux",
0014 reason="FPE monitoring currently only supported on Linux",
0015 ),
0016 pytest.mark.skipif(
0017 "ACTS_SEQUENCER_DISABLE_FPEMON" in os.environ,
0018 reason="Sequencer is configured to disable FPE monitoring",
0019 ),
0020 ]
0021
0022
0023 _names = {
0024 acts.FpeType.FLTDIV: "DivByZero",
0025 acts.FpeType.FLTOVF: "Overflow",
0026 acts.FpeType.FLTINV: "Invalid",
0027 }
0028
0029
0030 _types = [
0031 pytest.param(acts.FpeType.FLTDIV, id="FLTDIV"),
0032 pytest.param(acts.FpeType.FLTOVF, id="FLTOVF"),
0033 pytest.param(acts.FpeType.FLTINV, id="FLTINV"),
0034 ]
0035
0036 _src = (Path(__file__).parent / "../src/Framework.cpp").resolve()
0037 _locs = {}
0038 with _src.open() as fh:
0039 _name_to_type = {v.lower(): k for k, v in _names.items()}
0040 for i, line in enumerate(fh):
0041 m = re.match(r".*// ?MARK: (.*)", line)
0042 if m is None:
0043 continue
0044 (name,) = m.groups()
0045 _locs[_name_to_type[name]] = (str(_src), (i + 1, i + 2))
0046
0047
0048 class FpeMaker(acts.examples.IAlgorithm):
0049 def __init__(self, name):
0050 acts.examples.IAlgorithm.__init__(self, name, acts.logging.INFO)
0051
0052 def execute(self, context):
0053 i = context.eventNumber % 4
0054
0055 if i == 0 or i == 1:
0056 acts.FpeMonitor._trigger_divbyzero()
0057 elif i == 2:
0058 acts.FpeMonitor._trigger_overflow()
0059 elif i == 3:
0060 acts.FpeMonitor._trigger_invalid()
0061
0062 return acts.examples.ProcessCode.SUCCESS
0063
0064
0065 class FuncAlg(acts.examples.IAlgorithm):
0066 def __init__(self, name, func):
0067 acts.examples.IAlgorithm.__init__(self, name, acts.logging.INFO)
0068 self.func = func
0069
0070 def execute(self, context):
0071 self.func(context)
0072 return acts.examples.ProcessCode.SUCCESS
0073
0074
0075 @pytest.fixture(autouse=True)
0076 def disable_log_threshold():
0077 prev = acts.logging.getFailureThreshold()
0078 acts.logging.setFailureThreshold(acts.logging.MAX)
0079 yield
0080 acts.logging.setFailureThreshold(prev)
0081
0082
0083 def test_notrackfpe():
0084 s = acts.examples.Sequencer(
0085 events=3 * 100,
0086 trackFpes=False,
0087 )
0088 s.addAlgorithm(FpeMaker("FpeMaker"))
0089
0090 s.run()
0091
0092 res = s.fpeResult
0093
0094 for x in acts.FpeType.values:
0095 assert res.count(x) == 0
0096
0097
0098 @pytest.fixture(params=_types)
0099 def fpe_type(request):
0100 yield request.param
0101
0102
0103 def test_fpe_single_fail_at_end(fpe_type):
0104 s = acts.examples.Sequencer(
0105 events=10,
0106 failOnFirstFpe=False,
0107 )
0108
0109 s.addAlgorithm(
0110 FuncAlg(
0111 _names[fpe_type],
0112 lambda _: getattr(
0113 acts.FpeMonitor, f"_trigger_{_names[fpe_type].lower()}"
0114 )(),
0115 )
0116 )
0117 with pytest.raises(RuntimeError):
0118 s.run()
0119
0120 res = s.fpeResult
0121 for x in acts.FpeType.values:
0122 assert res.count(x) == (s.config.events if x == fpe_type else 0)
0123
0124
0125 def test_fpe_single_fail_immediately(fpe_type):
0126 s = acts.examples.Sequencer(
0127 events=10,
0128 failOnFirstFpe=True,
0129 numThreads=1,
0130 )
0131
0132 s.addAlgorithm(
0133 FuncAlg(
0134 _names[fpe_type],
0135 lambda _: getattr(
0136 acts.FpeMonitor, f"_trigger_{_names[fpe_type].lower()}"
0137 )(),
0138 )
0139 )
0140
0141 with pytest.raises(acts.FpeFailure):
0142 s.run()
0143
0144 res = s.fpeResult
0145 for x in acts.FpeType.values:
0146 assert res.count(x) == (1 if x == fpe_type else 0)
0147
0148
0149 def test_fpe_nocontext():
0150 class Alg(acts.examples.IAlgorithm):
0151 def __init__(self):
0152 acts.examples.IAlgorithm.__init__(self, "Alg", acts.logging.INFO)
0153
0154 def execute(self, context):
0155 assert context.fpeMonitor is None
0156 return acts.examples.ProcessCode.SUCCESS
0157
0158 s = acts.examples.Sequencer(
0159 events=10,
0160 trackFpes=False,
0161 numThreads=-1,
0162 )
0163 s.addAlgorithm(Alg())
0164 s.run()
0165
0166
0167 def test_fpe_rearm(fpe_type):
0168 trigger = getattr(acts.FpeMonitor, f"_trigger_{_names[fpe_type].lower()}")
0169
0170 class Alg(acts.examples.IAlgorithm):
0171 def __init__(self):
0172 acts.examples.IAlgorithm.__init__(self, "Alg", acts.logging.INFO)
0173
0174 def execute(self, context):
0175 assert context.fpeMonitor is not None
0176 trigger()
0177 context.fpeMonitor.rearm()
0178 trigger()
0179 return acts.examples.ProcessCode.SUCCESS
0180
0181 s = acts.examples.Sequencer(
0182 events=10,
0183 failOnFirstFpe=False,
0184 numThreads=-1,
0185 )
0186 s.addAlgorithm(Alg())
0187 with pytest.raises(RuntimeError):
0188 s.run()
0189
0190 res = s.fpeResult
0191 for x in acts.FpeType.values:
0192 assert res.count(x) == (s.config.events * 2 if x == fpe_type else 0)
0193
0194
0195 def test_fpe_masking_single(fpe_type):
0196 trigger = getattr(acts.FpeMonitor, f"_trigger_{_names[fpe_type].lower()}")
0197
0198 def func(context):
0199 trigger()
0200 assert context.fpeMonitor is not None
0201 context.fpeMonitor.rearm()
0202 trigger()
0203
0204
0205
0206 s = acts.examples.Sequencer(
0207 events=10,
0208 numThreads=-1,
0209 failOnFirstFpe=True,
0210 fpeMasks=[
0211 acts.examples.Sequencer.FpeMask(*_locs[fpe_type], fpe_type, 1),
0212 ],
0213 )
0214
0215 s.addAlgorithm(FuncAlg("Alg", func))
0216
0217 with pytest.raises(acts.FpeFailure):
0218 s.run()
0219
0220
0221
0222 s = acts.examples.Sequencer(
0223 events=10,
0224 numThreads=-1,
0225 failOnFirstFpe=True,
0226 fpeMasks=[
0227 acts.examples.Sequencer.FpeMask(*_locs[fpe_type], fpe_type, 3),
0228 ],
0229 )
0230
0231 s.addAlgorithm(FuncAlg("Alg", func))
0232
0233 s.run()
0234
0235 res = s.fpeResult
0236 for x in acts.FpeType.values:
0237 assert res.count(x) == (s.config.events * 2 if x == fpe_type else 0)
0238
0239
0240 def test_masking_load_yaml(fpe_type, tmp_path, monkeypatch):
0241 def eq(self, other):
0242 return (
0243 self.file == other.file
0244 and self.lines == other.lines
0245 and self.type == other.type
0246 and self.count == other.count
0247 )
0248
0249 monkeypatch.setattr(acts.examples.Sequencer.FpeMask, "__eq__", eq)
0250
0251 import yaml
0252
0253 masks = [
0254 acts.examples.Sequencer.FpeMask(*_locs[fpe_type], fpe_type, 1),
0255 ]
0256 file = tmp_path / "fpe_mask.yml"
0257 with file.open("w") as fh:
0258 yaml.dump(acts.examples.Sequencer.FpeMask.toDict(masks), fh)
0259
0260 masks2 = acts.examples.Sequencer.FpeMask.fromFile(file)
0261
0262 assert masks2 == masks
0263
0264
0265 def test_fpe_context(fpe_type):
0266 trigger = getattr(acts.FpeMonitor, f"_trigger_{_names[fpe_type].lower()}")
0267 trigger()
0268
0269 with acts.FpeMonitor.context() as fpe:
0270 trigger()
0271
0272 print(fpe.result)
0273
0274
0275 def test_buffer_sufficient():
0276 s = acts.examples.Sequencer(
0277 events=10000,
0278 failOnFirstFpe=False,
0279 )
0280
0281 s.addAlgorithm(FuncAlg("Invalid", lambda _: acts.FpeMonitor._trigger_invalid()))
0282 with pytest.raises(RuntimeError):
0283 s.run()
0284
0285 res = s.fpeResult
0286 for x in acts.FpeType.values:
0287 assert res.count(x) == (s.config.events if x == acts.FpeType.FLTINV else 0)
