will
25592fa6c2
ci / test (push) Successful in 6m14s
History-free snapshot of the private development repository at 77a82eb. Code, docs and the public catalogue only: no plant or job knowledge layers, no client, job or producer names (tools/scrub.py clean against the widened term list; a second word-level grep clean). The intro deck is left out because its slide images cannot be checked for job drawings.
510 lines
20 KiB
Python
510 lines
20 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests for `tools/deckcheck.py`.
|
|
|
|
Loaded by path, like `test_knowledge_drift.py` and `test_snapshot_defer.py`:
|
|
this is host CPython code sitting in a repository whose extension is
|
|
IronPython 2.7, and `conftest.py` in this folder registers the bare `vibe`
|
|
stub every one of these imports relies on.
|
|
|
|
The regression this exists to catch, spelled out as a test rather than left
|
|
as a paragraph: incident 1 in `tools/deckcheck.py`'s own docstring --
|
|
`test_regression_catalogue_value_changed_after_export_is_named_stale` builds
|
|
a deck frozen against an OLD catalogue rating, changes the catalogue the way
|
|
the 2026-09-01 P76 correction did, and asserts the STALE finding names the
|
|
mark and the field rather than passing silently.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
|
|
os.path.abspath(__file__))))
|
|
TOOLS = os.path.join(ROOT, 'tools')
|
|
if TOOLS not in sys.path:
|
|
sys.path.insert(0, TOOLS)
|
|
|
|
from vibe import knowledge # noqa: E402 (registers the vibe stub)
|
|
from vibe import plantpath # noqa: E402
|
|
|
|
deckcheck = pytest.importorskip('deckcheck')
|
|
D = deckcheck
|
|
|
|
|
|
# ------------------------------------------------------------------ helpers
|
|
|
|
def _write_yaml(path, data):
|
|
if not os.path.isdir(os.path.dirname(path)):
|
|
os.makedirs(os.path.dirname(path))
|
|
with io.open(path, 'w', encoding='utf-8') as fh:
|
|
yaml.safe_dump(data, fh)
|
|
|
|
|
|
def _write_json(path, data):
|
|
if not os.path.isdir(os.path.dirname(path)):
|
|
os.makedirs(os.path.dirname(path))
|
|
with io.open(path, 'w', encoding='utf-8') as fh:
|
|
json.dump(data, fh)
|
|
|
|
|
|
def _lifter(mark='M1', family='TESTLIFT', **extra):
|
|
row = {'mark': mark, 'family': family, 'x': 0.0, 'y': 0.0,
|
|
'pinned': False, 'pinned_y': False, 'on_edge': False,
|
|
'capacity_lb': None, 'shear_capacity_lb': None,
|
|
'basis_fc_psi': None, 'min_edge_distance_in': None,
|
|
'tension_fos': None, 'shear_fos': None}
|
|
row.update(extra)
|
|
return row
|
|
|
|
|
|
def _deck(mark='WP1', lifters=None, stages=None):
|
|
return {
|
|
'schema': 'panellift/1',
|
|
'panel': {'mark': mark, 'lifters': lifters or []},
|
|
'stages': stages or [],
|
|
'source': {},
|
|
}
|
|
|
|
|
|
class Repo(object):
|
|
"""A fixture knowledge tree: catalogue + plant + job layers on disk,
|
|
plus the job_root deckcheck should be told (or made to guess) about.
|
|
|
|
`plant_env` points WILLTOOLS_KNOWLEDGE at `<tmp>/plantroot/knowledge.yaml`
|
|
(a FILE, exactly as the real machine this was built on has it set --
|
|
`plantpath.plant_root()` reads that as "its directory is the root"), so
|
|
`plantpath.plant_knowledge_path('testplant')` resolves to
|
|
`<tmp>/plantroot/testplant/knowledge.yaml` with no other wiring needed.
|
|
"""
|
|
|
|
def __init__(self, tmp_path, monkeypatch, plant='testplant'):
|
|
self.tmp = tmp_path
|
|
self.job_root = str(tmp_path / 'job')
|
|
self.jobs_dir = os.path.join(self.job_root, 'sweep', 'jobs')
|
|
os.makedirs(self.jobs_dir)
|
|
self.catalogue_path = str(tmp_path / 'catalogue.yaml')
|
|
self.plant_path = str(tmp_path / 'plantroot' / plant / 'knowledge.yaml')
|
|
self.job_path = os.path.join(self.job_root, 'knowledge.yaml')
|
|
self.plant = plant
|
|
|
|
monkeypatch.setenv('WILLTOOLS_KNOWLEDGE',
|
|
str(tmp_path / 'plantroot' / 'knowledge.yaml'))
|
|
monkeypatch.setattr(D.knowledge, 'catalogue_path',
|
|
lambda: self.catalogue_path)
|
|
|
|
_write_yaml(self.catalogue_path, {'catalogue': {'inserts': {}}})
|
|
_write_yaml(self.job_path, {'job': {'plant': plant}})
|
|
_write_yaml(self.plant_path, {})
|
|
|
|
def set_catalogue(self, inserts):
|
|
# catalogue.yaml wraps its data in a top-level `catalogue:` key --
|
|
# `knowledge.get(know, 'catalogue.inserts', ...)` is what every
|
|
# lookup reads, so the fixture file has to nest the same way the
|
|
# real one does (see knowledge/catalogue.yaml itself, `catalogue:`
|
|
# at top level, `inserts:` inside it).
|
|
_write_yaml(self.catalogue_path, {'catalogue': {'inserts': inserts}})
|
|
|
|
def set_plant(self, data):
|
|
_write_yaml(self.plant_path, data)
|
|
|
|
def set_job(self, data):
|
|
data = dict(data)
|
|
data['job'] = {'plant': self.plant}
|
|
_write_yaml(self.job_path, data)
|
|
|
|
def deck_path(self, name):
|
|
return os.path.join(self.jobs_dir, name)
|
|
|
|
def build(self):
|
|
return D.build_knowledge(self.job_root)
|
|
|
|
|
|
@pytest.fixture
|
|
def repo(tmp_path, monkeypatch):
|
|
return Repo(tmp_path, monkeypatch)
|
|
|
|
|
|
# --------------------------------------------------------- build_knowledge
|
|
|
|
def test_layers_merge_defaults_catalogue_plant_job_in_that_order(repo):
|
|
repo.set_catalogue({'inserts': {}}) # catalogue: no override
|
|
repo.set_plant({'handling': {'tension_limit_sqrt_fc': 6.0}})
|
|
repo.set_job({'handling': {'tension_limit_sqrt_fc': 9.0}}) # job wins
|
|
know, layers = repo.build()
|
|
assert knowledge.get(know, 'handling.tension_limit_sqrt_fc') == 9.0
|
|
assert layers.plant_id == 'testplant'
|
|
assert layers.plant_status == 'merged'
|
|
assert layers.job_status == 'merged'
|
|
assert layers.catalogue_status == 'merged'
|
|
|
|
|
|
def test_plant_layer_alone_is_read_when_job_states_none(repo):
|
|
repo.set_plant({'handling': {'tension_limit_sqrt_fc': 6.0}})
|
|
_write_yaml(repo.job_path, {}) # no job.plant at all
|
|
know, layers = repo.build()
|
|
assert layers.plant_id is None
|
|
# Falls back to the MACHINE-WIDE file (<plant_root>/knowledge.yaml), not
|
|
# the per-plant one this fixture wrote -- which was never created, so
|
|
# this is the ordinary "no plant file yet" case and stays silent.
|
|
assert layers.plant_status == 'absent'
|
|
assert knowledge.get(know, 'handling.tension_limit_sqrt_fc') == 5.0 # DEFAULTS
|
|
|
|
|
|
def test_named_plant_with_no_folder_is_missing_not_silently_skipped(repo):
|
|
_write_yaml(repo.job_path, {'job': {'plant': 'nobody_configured_this'}})
|
|
know, layers = repo.build()
|
|
assert layers.plant_status == 'missing'
|
|
assert layers.plant_id == 'nobody_configured_this'
|
|
|
|
|
|
def test_unreadable_yaml_layer_is_recorded_and_skipped_whole(repo, tmp_path):
|
|
bad = tmp_path / 'bad.yaml'
|
|
bad.write_text('a: [1, 2\n') # malformed
|
|
status = D._merge_layer({}, str(bad))
|
|
assert status == 'unreadable'
|
|
|
|
|
|
def test_absent_layer_is_silent(repo, tmp_path):
|
|
status = D._merge_layer({}, str(tmp_path / 'does-not-exist.yaml'))
|
|
assert status == 'absent'
|
|
|
|
|
|
# ------------------------------------------------------------- _guess_job_root
|
|
|
|
def test_guess_job_root_finds_the_folder_holding_knowledge_yaml(repo):
|
|
guessed = D._guess_job_root(repo.jobs_dir)
|
|
assert os.path.normcase(guessed) == os.path.normcase(repo.job_root)
|
|
|
|
|
|
# -------------------------------------------------------------- _num_eq / _verdict
|
|
|
|
def test_num_eq_treats_both_none_as_equal():
|
|
assert D._num_eq(None, None) is True
|
|
|
|
|
|
def test_num_eq_one_none_is_not_equal():
|
|
assert D._num_eq(None, 5) is False
|
|
assert D._num_eq(5, None) is False
|
|
|
|
|
|
def test_verdict_both_none_is_current():
|
|
assert D._verdict(None, None) == D.CURRENT
|
|
|
|
|
|
def test_verdict_current_none_frozen_value_is_unjudgeable():
|
|
assert D._verdict(None, 7690.0) == D.UNJUDGEABLE
|
|
|
|
|
|
def test_verdict_frozen_none_current_value_is_stale():
|
|
assert D._verdict(16000.0, None) == D.STALE
|
|
|
|
|
|
def test_verdict_matching_numbers_is_current():
|
|
assert D._verdict(16000.0, 16000.0) == D.CURRENT
|
|
|
|
|
|
def test_verdict_mismatched_numbers_is_stale():
|
|
assert D._verdict(16000.0, 7690.0) == D.STALE
|
|
|
|
|
|
# ---------------------------------------------------- check_capacity: current
|
|
|
|
def test_capacity_current_for_a_plain_ungated_family(repo):
|
|
repo.set_catalogue({'PLAIN01': {'working_load_lb': 8000,
|
|
'shear_load_lb': 8000}})
|
|
repo.set_plant({'lifters': {'catalogue_alias': {'PLAINLIFT': 'PLAIN01'}}})
|
|
know, _ = repo.build()
|
|
status, detail = D.check_capacity(know, 'PLAINLIFT', 'tension', 8000.0, [])
|
|
assert status == D.CURRENT
|
|
assert detail['current'] == 8000.0
|
|
|
|
|
|
# ----------------------------------------------- check_capacity: the regression
|
|
|
|
def test_regression_catalogue_value_changed_after_export_is_named_stale(repo):
|
|
"""Incident 1, reproduced: a deck was frozen against a published rating
|
|
that the catalogue later corrected. The mismatch must be named --
|
|
mark, field, frozen and current -- not merged away or missed.
|
|
"""
|
|
repo.set_catalogue({'P76ALIAS': {'working_load_lb': 7690}})
|
|
repo.set_plant({'lifters': {'catalogue_alias': {'LIFTFAM_P76': 'P76ALIAS'}}})
|
|
know_old, _ = repo.build()
|
|
status_old, detail_old = D.check_capacity(know_old, 'LIFTFAM_P76',
|
|
'tension', 7690.0, [])
|
|
assert status_old == D.CURRENT # the deck agreed with knowledge AT EXPORT
|
|
|
|
# The correction: 7690 -> 16000, same shape as the real P76 fix.
|
|
repo.set_catalogue({'P76ALIAS': {'working_load_lb': 16000}})
|
|
know_new, _ = repo.build()
|
|
status_new, detail_new = D.check_capacity(know_new, 'LIFTFAM_P76',
|
|
'tension', 7690.0, [])
|
|
assert status_new == D.STALE
|
|
assert detail_new['frozen'] == 7690.0
|
|
assert detail_new['current'] == 16000.0
|
|
|
|
# And end-to-end through check_deck(), which is what names the mark.
|
|
deck_path = repo.deck_path('WP900.job.json')
|
|
_write_json(deck_path, _deck('WP900', lifters=[
|
|
_lifter('L1', 'LIFTFAM_P76', capacity_lb=7690.0)]))
|
|
findings = D.check_deck(deck_path, know_new)
|
|
hit = [f for f in findings if f['field'] == 'capacity_lb']
|
|
assert len(hit) == 1
|
|
assert hit[0]['status'] == D.STALE
|
|
assert hit[0]['mark'] == 'WP900'
|
|
assert hit[0]['detail']['frozen'] == 7690.0
|
|
assert hit[0]['detail']['current'] == 16000.0
|
|
|
|
|
|
# --------------------------------------------------------- check_capacity: gated
|
|
|
|
def test_capacity_conditional_when_tension_bar_gate_unresolvable_offline(repo):
|
|
repo.set_catalogue({'GATED01': {
|
|
'working_load_lb': 16000, 'shear_load_lb': 4010,
|
|
'requires_tension_bar_for_rating': True,
|
|
'tension_without_tension_bar_lb': 7690,
|
|
}})
|
|
repo.set_plant({'lifters': {
|
|
'catalogue_alias': {'GATEDFAM_P76': 'GATED01'},
|
|
'tension_bar_families': ['SR5602'],
|
|
'parts_shipped_with_tension_bar': ['P76'],
|
|
}})
|
|
know, _ = repo.build()
|
|
bar_families = knowledge.get(know, 'lifters.tension_bar_families', [])
|
|
|
|
# frozen matches the WITHOUT-bar figure -- the case that must NOT read
|
|
# as a quiet 'current' match, because the true answer depends on
|
|
# assembly context this call was never given.
|
|
status, detail = D.check_capacity(know, 'GATEDFAM_P76', 'tension',
|
|
7690.0, bar_families)
|
|
assert status == D.CONDITIONAL
|
|
assert detail['without_bar'] == 7690.0
|
|
assert detail['with_bar'] == 16000.0
|
|
assert detail['frozen_matches'] == ['without_bar']
|
|
|
|
# frozen matches the WITH-bar figure -- still conditional, never
|
|
# silently 'current' either, and frozen_matches names the other side.
|
|
status2, detail2 = D.check_capacity(know, 'GATEDFAM_P76', 'tension',
|
|
16000.0, bar_families)
|
|
assert status2 == D.CONDITIONAL
|
|
assert detail2['frozen_matches'] == ['with_bar']
|
|
|
|
|
|
def test_capacity_shear_is_never_gated(repo):
|
|
# The gate in `knowledge.lifter_capacity` only applies to kind='tension'
|
|
# -- shear must resolve deterministically even for a gated family.
|
|
repo.set_catalogue({'GATED01': {
|
|
'working_load_lb': 16000, 'shear_load_lb': 4010,
|
|
'requires_tension_bar_for_rating': True,
|
|
'tension_without_tension_bar_lb': 7690,
|
|
}})
|
|
repo.set_plant({'lifters': {
|
|
'catalogue_alias': {'GATEDFAM_P76': 'GATED01'},
|
|
'tension_bar_families': ['SR5602'],
|
|
'parts_shipped_with_tension_bar': ['P76'],
|
|
}})
|
|
know, _ = repo.build()
|
|
status, detail = D.check_capacity(know, 'GATEDFAM_P76', 'shear', 4010.0, [])
|
|
assert status == D.CURRENT
|
|
assert detail['current'] == 4010.0
|
|
|
|
|
|
def test_capacity_unjudgeable_when_no_catalogue_alias(repo):
|
|
know, _ = repo.build()
|
|
status, detail = D.check_capacity(know, 'NOBODY_KNOWS_THIS', 'tension',
|
|
7690.0, [])
|
|
assert status == D.UNJUDGEABLE
|
|
assert 'reason' not in detail or detail.get('current') is None
|
|
|
|
|
|
# ------------------------------------------------------------- stage checks
|
|
|
|
def test_stage_tension_limit_current_when_job_wide_matches(repo):
|
|
repo.set_plant({'handling': {'tension_limit_sqrt_fc': 7.5}})
|
|
know, _ = repo.build()
|
|
stage = {'name': 'stripping', 'pci_case': 'stripping',
|
|
'tension_limit_sqrt_fc': 7.5, 'tension_limit_source': 'job'}
|
|
status, detail = D.check_stage_tension_limit(
|
|
know, stage, knowledge.get(know, 'handling.tension_limit_sqrt_fc', 5.0))
|
|
assert status == D.CURRENT
|
|
|
|
|
|
def test_stage_tension_limit_stale_when_job_wide_moved(repo):
|
|
repo.set_plant({'handling': {'tension_limit_sqrt_fc': 5.0}})
|
|
know, _ = repo.build()
|
|
stage = {'name': 'stripping', 'pci_case': 'stripping',
|
|
'tension_limit_sqrt_fc': 7.5, 'tension_limit_source': 'job'}
|
|
status, detail = D.check_stage_tension_limit(
|
|
know, stage, knowledge.get(know, 'handling.tension_limit_sqrt_fc', 5.0))
|
|
assert status == D.STALE
|
|
assert detail['frozen'] == 7.5
|
|
assert detail['current'] == 5.0
|
|
|
|
|
|
def test_stage_tension_limit_unjudgeable_when_no_current_spec_matches(repo):
|
|
repo.set_plant({'handling': {'stages': []}}) # the stage was removed
|
|
know, _ = repo.build()
|
|
stage = {'name': 'transportation', 'pci_case': 'transportation',
|
|
'tension_limit_sqrt_fc': 9.0, 'tension_limit_source': 'stage'}
|
|
status, detail = D.check_stage_tension_limit(know, stage, 5.0)
|
|
assert status == D.UNJUDGEABLE
|
|
|
|
|
|
# --------------------------------------------------------------- check_deck
|
|
|
|
def test_check_deck_skips_fields_the_deck_never_recorded(repo):
|
|
repo.set_catalogue({'PLAIN01': {'working_load_lb': 8000}})
|
|
repo.set_plant({'lifters': {'catalogue_alias': {'PLAINLIFT': 'PLAIN01'}}})
|
|
know, _ = repo.build()
|
|
# A minimal, panellift/1-shaped lifter with only capacity_lb recorded --
|
|
# no basis_fc_psi, no fos keys. Those fields must not appear as findings
|
|
# at all: absent is not the same as a mismatch.
|
|
deck_path = repo.deck_path('WP1.job.json')
|
|
_write_json(deck_path, _deck('WP1', lifters=[
|
|
{'mark': 'L1', 'family': 'PLAINLIFT', 'capacity_lb': 8000.0}]))
|
|
findings = D.check_deck(deck_path, know)
|
|
fields = set(f['field'] for f in findings)
|
|
assert fields == {'capacity_lb'}
|
|
|
|
|
|
def test_check_deck_lifter_with_no_family_is_unjudgeable(repo):
|
|
know, _ = repo.build()
|
|
deck_path = repo.deck_path('WP2.job.json')
|
|
_write_json(deck_path, _deck('WP2', lifters=[
|
|
{'mark': 'L1', 'family': None, 'capacity_lb': 100.0}]))
|
|
findings = D.check_deck(deck_path, know)
|
|
assert len(findings) == 1
|
|
assert findings[0]['status'] == D.UNJUDGEABLE
|
|
assert findings[0]['field'] == 'family'
|
|
|
|
|
|
def test_deck_status_is_the_worst_finding():
|
|
findings = [
|
|
{'status': D.CURRENT}, {'status': D.UNJUDGEABLE},
|
|
{'status': D.CONDITIONAL},
|
|
]
|
|
assert D.deck_status(findings) == D.CONDITIONAL
|
|
findings.append({'status': D.STALE})
|
|
assert D.deck_status(findings) == D.STALE
|
|
assert D.deck_status([]) == D.CURRENT
|
|
|
|
|
|
# -------------------------------------------------------------- _deck_lines
|
|
|
|
def test_deck_lines_group_identical_findings_across_lifters(repo):
|
|
findings = [
|
|
{'deck': 'x', 'mark': 'WP1', 'lifter_mark': 'L1', 'family': 'F',
|
|
'field': 'capacity_lb', 'status': D.STALE,
|
|
'detail': {'frozen': 7690.0, 'current': 16000.0}},
|
|
{'deck': 'x', 'mark': 'WP1', 'lifter_mark': 'L2', 'family': 'F',
|
|
'field': 'capacity_lb', 'status': D.STALE,
|
|
'detail': {'frozen': 7690.0, 'current': 16000.0}},
|
|
]
|
|
lines = D._deck_lines('WP1', findings)
|
|
assert len(lines) == 1
|
|
assert '2 lifters' in lines[0]
|
|
assert 'WP1' in lines[0]
|
|
|
|
|
|
def test_deck_lines_keep_genuinely_different_values_separate():
|
|
findings = [
|
|
{'deck': 'x', 'mark': 'WP1', 'lifter_mark': 'L1', 'family': 'F',
|
|
'field': 'capacity_lb', 'status': D.STALE,
|
|
'detail': {'frozen': 7690.0, 'current': 16000.0}},
|
|
{'deck': 'x', 'mark': 'WP1', 'lifter_mark': 'L2', 'family': 'F',
|
|
'field': 'capacity_lb', 'status': D.STALE,
|
|
'detail': {'frozen': 8000.0, 'current': 9000.0}},
|
|
]
|
|
lines = D._deck_lines('WP1', findings)
|
|
assert len(lines) == 2
|
|
|
|
|
|
def test_deck_lines_omit_current_findings():
|
|
findings = [{'deck': 'x', 'mark': 'WP1', 'lifter_mark': 'L1',
|
|
'family': 'F', 'field': 'capacity_lb', 'status': D.CURRENT,
|
|
'detail': {'current': 1.0, 'frozen': 1.0}}]
|
|
assert D._deck_lines('WP1', findings) == []
|
|
|
|
|
|
# --------------------------------------------------------------------- CLI
|
|
|
|
def test_main_exits_1_when_a_deck_is_stale(repo, capsys):
|
|
repo.set_catalogue({'P76ALIAS': {'working_load_lb': 16000}})
|
|
repo.set_plant({'lifters': {'catalogue_alias': {'LIFTFAM_P76': 'P76ALIAS'}}})
|
|
_write_json(repo.deck_path('WP1.job.json'), _deck('WP1', lifters=[
|
|
_lifter('L1', 'LIFTFAM_P76', capacity_lb=7690.0)]))
|
|
rc = D.main([repo.jobs_dir])
|
|
out = capsys.readouterr().out
|
|
assert rc == 1
|
|
assert 'STALE' in out
|
|
assert 'WP1' in out
|
|
assert 'capacity_lb' in out
|
|
|
|
|
|
def test_main_exits_0_when_everything_is_current(repo, capsys):
|
|
repo.set_catalogue({'PLAIN01': {'working_load_lb': 8000}})
|
|
repo.set_plant({'lifters': {'catalogue_alias': {'PLAINLIFT': 'PLAIN01'}}})
|
|
_write_json(repo.deck_path('WP1.job.json'), _deck('WP1', lifters=[
|
|
_lifter('L1', 'PLAINLIFT', capacity_lb=8000.0)]))
|
|
rc = D.main([repo.jobs_dir])
|
|
out = capsys.readouterr().out
|
|
assert rc == 0
|
|
assert '1 deck(s) checked, 1 current, 0 stale, 0 conditional, 0 unjudgeable' in out
|
|
|
|
|
|
def test_main_exits_1_on_conditional_only(repo, capsys):
|
|
repo.set_catalogue({'GATED01': {
|
|
'working_load_lb': 16000, 'shear_load_lb': 4010,
|
|
'requires_tension_bar_for_rating': True,
|
|
'tension_without_tension_bar_lb': 7690,
|
|
}})
|
|
repo.set_plant({'lifters': {
|
|
'catalogue_alias': {'GATEDFAM_P76': 'GATED01'},
|
|
'tension_bar_families': ['SR5602'],
|
|
'parts_shipped_with_tension_bar': ['P76'],
|
|
}})
|
|
_write_json(repo.deck_path('WP1.job.json'), _deck('WP1', lifters=[
|
|
_lifter('L1', 'GATEDFAM_P76', capacity_lb=7690.0,
|
|
shear_capacity_lb=4010.0)]))
|
|
rc = D.main([repo.jobs_dir])
|
|
out = capsys.readouterr().out
|
|
assert rc == 1
|
|
assert 'CONDITIONAL' in out
|
|
|
|
|
|
def test_main_exits_2_when_jobs_dir_missing(tmp_path):
|
|
rc = D.main([str(tmp_path / 'nope')])
|
|
assert rc == 2
|
|
|
|
|
|
def test_main_exits_2_when_jobs_dir_has_no_decks(repo):
|
|
rc = D.main([repo.jobs_dir])
|
|
assert rc == 2
|
|
|
|
|
|
def test_main_prints_the_model_caveat_once(repo, capsys):
|
|
repo.set_catalogue({'PLAIN01': {'working_load_lb': 8000}})
|
|
repo.set_plant({'lifters': {'catalogue_alias': {'PLAINLIFT': 'PLAIN01'}}})
|
|
_write_json(repo.deck_path('WP1.job.json'), _deck('WP1', lifters=[
|
|
_lifter('L1', 'PLAINLIFT', capacity_lb=8000.0)]))
|
|
D.main([repo.jobs_dir])
|
|
out = capsys.readouterr().out
|
|
assert out.count('NOT against the live') == 1 or 'MODEL' in out
|
|
|
|
|
|
def test_main_writes_json_output(repo, tmp_path):
|
|
repo.set_catalogue({'PLAIN01': {'working_load_lb': 8000}})
|
|
repo.set_plant({'lifters': {'catalogue_alias': {'PLAINLIFT': 'PLAIN01'}}})
|
|
_write_json(repo.deck_path('WP1.job.json'), _deck('WP1', lifters=[
|
|
_lifter('L1', 'PLAINLIFT', capacity_lb=8000.0)]))
|
|
out_path = str(tmp_path / 'out.json')
|
|
D.main([repo.jobs_dir, '--json', out_path])
|
|
with io.open(out_path, encoding='utf-8') as fh:
|
|
payload = json.load(fh)
|
|
assert payload['decks_checked'] == 1
|
|
assert payload['current'] == 1
|