| self = <test_oracle_corpus_nightly_service.TestNightlyHotSwitch object at 0x15be4a9f0>
|
|
|
| def test_hot_switch_default_passes_through(self):
|
| """fail-OPEN: a missing flag must NOT silently halt the job. The setting's
|
| value is what hot_flag falls back to."""
|
| task = _get_task()
|
| seen = []
|
| def fake_hot_flag(name, default):
|
| seen.append((name, default))
|
| return default
|
| runner = MagicMock()
|
| fake_settings = MagicMock()
|
| fake_settings.ORACLE_CORPUS_NIGHTLY_ENABLED = True
|
| fake_settings.ORACLE_CORPUS_NIGHTLY_TRAILING_DAYS = 30
|
| with patch("rtrader.services.feature_flags.hot_flag", side_effect=fake_hot_flag), \
|
| patch("rtrader.config.settings", fake_settings):
|
| > task(date(2026, 7, 24), _walk_runner=runner)
|
|
|
| tests/cli/test_oracle_corpus_nightly_service.py:97:
|
| _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
|
| rtrader/cli/services_cli.py:2419: in _oracle_corpus_nightly_task
|
| history(chunk_months=int(max(1, chunk_months)))
|
| rtrader/cli/services_cli.py:2446: in _default_oracle_corpus_history_runner
|
| _run_market_turn_cli(
|
| rtrader/cli/services_cli.py:2319: in _run_market_turn_cli
|
| proc = subprocess.run(cmd, capture_output=True, text=True)
|
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
| ../.pyenv/versions/3.12.12/lib/python3.12/subprocess.py:550: in run
|
| stdout, stderr = process.communicate(input, timeout=timeout)
|
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
| ../.pyenv/versions/3.12.12/lib/python3.12/subprocess.py:1209: in communicate
|
| stdout, stderr = self._communicate(input, endtime, timeout)
|
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
| ../.pyenv/versions/3.12.12/lib/python3.12/subprocess.py:2115: in _communicate
|
| ready = selector.select(timeout)
|
| ^^^^^^^^^^^^^^^^^^^^^^^^
|
| _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
|
|
|
| self = <selectors.PollSelector object at 0x15d80f4a0>, timeout = None
|
|
|
| def select(self, timeout=None):
|
| # This is shared between poll() and epoll().
|
| # epoll() has a different signature and handling of timeout parameter.
|
| if timeout is None:
|
| timeout = None
|
| elif timeout <= 0:
|
| timeout = 0
|
| else:
|
| # poll() has a resolution of 1 millisecond, round away from
|
| # zero to wait *at least* timeout seconds.
|
| timeout = math.ceil(timeout * 1e3)
|
| ready = []
|
| try:
|
| > fd_event_list = self._selector.poll(timeout)
|
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
| E Failed: Timeout (>30.0s) from pytest-timeout.
|
|
|
| ../.pyenv/versions/3.12.12/lib/python3.12/selectors.py:415: Failed
|