check formatting with YARF

Author: impself

tests/tools/safety/conftest.py

@@ -31,12 +31,16 @@
 def _policy_dict(**overrides: Any) -> dict[str, Any]:
     data: dict[str, Any] = {
         "version": POLICY_VERSION,
-        "network": {"allow_domains": ["api.example.com", "*.example.com"]},
+        "network": {
+            "allow_domains": ["api.example.com", "*.example.com"]
+        },
         "commands": {
             "allow": ["python", "ls"],
             "deny": ["rm"],
         },
-        "paths": {"deny": ["/etc/**", "~/.ssh/**", "/root/**"]},
+        "paths": {
+            "deny": ["/etc/**", "~/.ssh/**", "/root/**"]
+        },
         "limits": {
             "max_timeout_seconds": 60.0,
             "max_output_bytes": 1024,
@@ -51,8 +55,13 @@ def _policy_dict(**overrides: Any) -> dict[str, Any]:
             "guard_error": "deny",
             "human_review_blocks_execution": True,
         },
-        "dependencies": {"decision": "deny"},
-        "audit": {"enabled": False, "required": False},
+        "dependencies": {
+            "decision": "deny"
+        },
+        "audit": {
+            "enabled": False,
+            "required": False
+        },
     }
     for key, value in overrides.items():
         data[key] = value

tests/tools/safety/test_audit.py

@@ -32,7 +32,7 @@ def _make_event() -> SafetyAuditEvent:
         tool_kind=ToolKind.UNKNOWN,
         decision=SafetyDecision.DENY,
         risk_level=RiskLevel.HIGH,
-        rule_ids=("FILE001_RECURSIVE_DELETE",),
+        rule_ids=("FILE001_RECURSIVE_DELETE", ),
         duration_ms=0.5,
         redacted=False,
         execution_blocked=True,
@@ -43,6 +43,7 @@ def _make_event() -> SafetyAuditEvent:
 
 
 class TestInMemoryAuditSink:
+
     def test_protocol(self):
         sink = InMemoryAuditSink()
         assert isinstance(sink, AuditSink)
@@ -65,16 +66,14 @@ async def test_clear(self):
     @pytest.mark.asyncio
     async def test_concurrent_emits_keep_order(self):
         sink = InMemoryAuditSink()
-        events = [
-            _make_event().model_copy(update={"event_id": f"e{i}"})
-            for i in range(5)
-        ]
+        events = [_make_event().model_copy(update={"event_id": f"e{i}"}) for i in range(5)]
         await asyncio.gather(*(sink.emit(e) for e in events))
         ids = [e.event_id for e in sink.events]
         assert sorted(ids) == ["e0", "e1", "e2", "e3", "e4"]
 
 
 class TestJsonlAuditSink:
+
     @pytest.mark.asyncio
     async def test_append_writes_line(self, tmp_path):
         path = tmp_path / "audit.jsonl"
@@ -101,6 +100,7 @@ def test_accepts_pathlike_and_str(self, tmp_path):
 
 
 class TestNullAuditSink:
+
     @pytest.mark.asyncio
     async def test_emit_noop(self):
         # Just verify it does not raise.

tests/tools/safety/test_bash_scanner.py

@@ -31,6 +31,7 @@ def _scan(src: str):
 
 
 class TestParseSleep:
+
     def test_plain_seconds(self):
         assert _parse_sleep("5") == 5.0
 
@@ -54,6 +55,7 @@ def test_invalid(self):
 
 
 class TestHostHelpers:
+
     def test_looks_like_host_simple(self):
         assert _looks_like_host("api.example.com") is True
 
@@ -72,6 +74,7 @@ def test_looks_like_host_path(self):
 
 
 class TestSecretRef:
+
     def test_token_var(self):
         assert _looks_like_secret_ref("${API_TOKEN}") is True
 
@@ -84,6 +87,7 @@ def test_plain_var(self):
 
 
 class TestEnvAssignment:
+
     def test_simple(self):
         assert _looks_like_env_assignment("FOO=bar") is True
 
@@ -101,6 +105,7 @@ def test_no_equals(self):
 
 
 class TestDdHelpers:
+
     def test_size_bs_count(self):
         assert _extract_dd_size(["bs=1M", "count=2"]) == 2 * 1024 * 1024
 
@@ -130,9 +135,9 @@ def test_parse_size_value_invalid(self):
 
 
 class TestPythonPipInstall:
+
     def test_python_m_pip_install(self):
-        assert _is_python_pip_install("python",
-                                       ["-m", "pip", "install", "x"]) is True
+        assert _is_python_pip_install("python", ["-m", "pip", "install", "x"]) is True
 
     def test_python_no_pip(self):
         assert _is_python_pip_install("python", ["script.py"]) is False
@@ -142,6 +147,7 @@ def test_non_python(self):
 
 
 class TestInterpreterPayload:
+
     def test_dash_c(self):
         assert _interpreter_runs_payload(["-c", "code"]) is True
 
@@ -156,6 +162,7 @@ def test_only_flags(self):
 
 
 class TestOffsets:
+
     def test_line_of_first_line(self):
         assert _line_of("hello\nworld", 1) == 1
 
@@ -172,6 +179,7 @@ def test_col_of_second_line(self):
 
 
 class TestRm:
+
     def test_recursive_rm(self):
         facts = _scan("rm -rf /tmp/x")
         assert facts.file_deletes
@@ -190,6 +198,7 @@ def test_rm_no_target(self):
 
 
 class TestNetwork:
+
     def test_curl_static_url(self):
         facts = _scan("curl https://api.example.com/x")
         assert facts.network_calls
@@ -217,11 +226,11 @@ def test_url_in_arg_catchall(self):
         # command follows. Use a plain command that survives to the URL
         # catch-all in the generic handler.
         facts = _scan("true https://evil.example.com")
-        assert any(n.target == "evil.example.com"
-                   for n in facts.network_calls)
+        assert any(n.target == "evil.example.com" for n in facts.network_calls)
 
 
 class TestFileRead:
+
     def test_cat_dotenv(self):
         facts = _scan("cat .env")
         assert facts.file_reads
@@ -234,6 +243,7 @@ def test_cat_ssh_key(self):
 
 
 class TestFileWrite:
+
     def test_tee(self):
         facts = _scan("echo hi | tee /tmp/x")
         assert any(w.target == "/tmp/x" for w in facts.file_writes)
@@ -245,13 +255,15 @@ def test_redirection(self):
 
 
 class TestPrivilege:
+
     def test_sudo(self):
         facts = _scan("sudo rm /tmp/x")
         assert facts.privilege_commands
         assert facts.privilege_commands[0].command == "sudo"
 
 
 class TestPackageManager:
+
     def test_pip_install(self):
         facts = _scan("pip install requests")
         assert facts.dependency_installs
@@ -273,6 +285,7 @@ def test_apt_install(self):
 
 
 class TestDynamicExec:
+
     def test_eval(self):
         facts = _scan("eval 'rm -rf /'")
         assert facts.dynamic_execs
@@ -308,6 +321,7 @@ def test_interpreter_payload(self):
 
 
 class TestShellOperators:
+
     def test_pipe(self):
         facts = _scan("a | b")
         assert any(o.operator == "|" for o in facts.shell_operators)
@@ -329,6 +343,7 @@ def test_semicolon(self):
 
 
 class TestSleep:
+
     def test_static_sleep(self):
         facts = _scan("sleep 5")
         assert facts.long_sleeps
@@ -341,6 +356,7 @@ def test_invalid_sleep(self):
 
 
 class TestLargeWrites:
+
     def test_dd_size_extracted(self):
         facts = _scan("dd if=/dev/zero of=/tmp/x bs=1M count=10")
         assert facts.large_writes
@@ -349,12 +365,14 @@ def test_dd_size_extracted(self):
 
 
 class TestForkBomb:
+
     def test_classic_bomb(self):
         facts = _scan(":(){ :|:& };:")
         assert facts.fork_bombs
 
 
 class TestLoops:
+
     def test_while_true(self):
         facts = _scan("while true; do echo hi; done")
         assert facts.unbounded_loops
@@ -365,6 +383,7 @@ def test_for_infinite(self):
 
 
 class TestConcurrency:
+
     def test_many_background_jobs(self):
         # 8+ single-amp background jobs trigger concurrency fact
         facts = _scan("a & b & c & d & e & f & g & h &")
@@ -373,6 +392,7 @@ def test_many_background_jobs(self):
 
 
 class TestSecretFlowBash:
+
     def test_echo_token(self):
         facts = _scan('echo "$API_TOKEN"')
         assert facts.secret_flows
@@ -384,6 +404,7 @@ def test_echo_password(self):
 
 
 class TestLexer:
+
     def test_unbalanced_quote(self):
         lexer = _BashLexer("'unterminated")
         _, errors = lexer.tokenize()
@@ -416,6 +437,7 @@ def test_newline_separator(self):
 
 
 class TestEnvAssignmentHandling:
+
     def test_leading_env_stripped(self):
         facts = _scan("FOO=bar ls -l")
         assert facts.process_calls
@@ -428,17 +450,16 @@ def test_env_only_no_command(self):
 
 
 class TestBashScannerRule:
+
     def test_skips_when_language_mismatch(self, scan_request_factory):
         rule = BashScannerRule()
-        req = scan_request_factory(
-            language=ScriptLanguage.PYTHON, script="print(1)")
+        req = scan_request_factory(language=ScriptLanguage.PYTHON, script="print(1)")
         out = list(rule.scan(req, _make_min_policy()))
         assert out == []
 
     def test_emits_finding_for_dangerous(self, scan_request_factory):
         rule = BashScannerRule()
-        req = scan_request_factory(
-            language=ScriptLanguage.BASH, script="rm -rf /tmp")
+        req = scan_request_factory(language=ScriptLanguage.BASH, script="rm -rf /tmp")
         out = list(rule.scan(req, _make_min_policy()))
         assert any(f.rule_id == "FILE001_RECURSIVE_DELETE" for f in out)