3
3
import json
4
4
import tempfile
5
5
6
+ import subprocess
6
7
from dataclasses import dataclass
7
8
from typing import Any , Dict , List , Optional
8
9
from pathlib import Path
9
10
10
- from click . testing import CliRunner
11
- from opentrons . cli . analyze import analyze
11
+ import pytest
12
+
12
13
from opentrons .protocols .api_support .definitions import MAX_SUPPORTED_VERSION
13
14
14
15
23
24
class _AnalysisCLIResult :
24
25
exit_code : int
25
26
json_output : Optional [Dict [str , Any ]]
26
- stdout_stderr : str
27
+ stdout_stderr : bytes
27
28
28
29
29
30
# Function copied from api/tests/opentrons/cli/test_cli.py
@@ -45,8 +46,14 @@ def _get_analysis_result(
45
46
"""
46
47
with tempfile .TemporaryDirectory () as temp_dir :
47
48
analysis_output_file = Path (temp_dir ) / "analysis_output.json"
48
- runner = CliRunner ()
49
- args = [output_type , str (analysis_output_file )]
49
+ args = [
50
+ "python" ,
51
+ "-m" ,
52
+ "opentrons.cli" ,
53
+ "analyze" ,
54
+ output_type ,
55
+ str (analysis_output_file ),
56
+ ]
50
57
51
58
if rtp_values is not None :
52
59
args .extend (["--rtp-values" , rtp_values ])
@@ -59,32 +66,46 @@ def _get_analysis_result(
59
66
if check :
60
67
args .append ("--check" )
61
68
62
- result = runner . invoke ( analyze , args )
69
+ process = subprocess . run ( args , stdout = subprocess . PIPE , stderr = subprocess . STDOUT )
63
70
if analysis_output_file .exists ():
64
71
json_output = json .loads (analysis_output_file .read_bytes ())
65
72
else :
66
73
json_output = None
67
74
return _AnalysisCLIResult (
68
- exit_code = result . exit_code ,
75
+ exit_code = process . returncode ,
69
76
json_output = json_output ,
70
- stdout_stderr = result . output ,
77
+ stdout_stderr = process . stdout ,
71
78
)
72
79
73
80
74
- # TODO (spp, 2025-06-10): update this test to verify that the gravimetric protocol
75
- # analyzes successfully once the protocol and CSV file values are corrected.
76
- # It's better to write a *single* test that verifies all aspects of its analysis result
77
- # than writing a different one for each aspect since the gravimetric test is complex
78
- # and analyzing it takes a long time.
79
- def test_gravimetric_test_protocol_uses_latest_api_version () -> None :
80
- """Should check that gravimetric test protocol uses the latest Python API version."""
81
+ @pytest .mark .parametrize (
82
+ "pipette" ,
83
+ [
84
+ pytest .param (
85
+ "96ch200" , marks = pytest .mark .xfail (reason = "200ul has no liquid class" )
86
+ ),
87
+ pytest .param ("96ch1000" ),
88
+ ],
89
+ )
90
+ def test_gravimetric_test_protocol_passes_analysis (pipette : str ) -> None :
91
+ """Check that gravimetric test protocol uses the latest Python API version and simulates."""
81
92
result = _get_analysis_result (
82
93
[GRAVIMETRIC_PROTOCOL_FILEPATH ],
83
94
"--json-output" ,
84
- rtp_files = json .dumps ({"qc_test_profile" : str (CSV_FILEPATH .resolve ())}),
95
+ rtp_files = json .dumps (
96
+ {
97
+ "qc_test_profile" : str (
98
+ (GRAVIMETRIC_PROTOCOL_PARENT_FILEPATH / f"{ pipette } .csv" ).resolve ()
99
+ )
100
+ }
101
+ ),
85
102
)
103
+ print (result .stdout_stderr )
86
104
assert result .exit_code == 0
87
- assert result .json_output is not None
105
+ assert result .json_output
106
+ assert result .json_output ["errors" ] == [], "Analysis failed: " + str (
107
+ result .json_output
108
+ )
88
109
assert result .json_output ["config" ]["apiVersion" ] == [
89
110
MAX_SUPPORTED_VERSION .major ,
90
111
MAX_SUPPORTED_VERSION .minor ,
0 commit comments