|
1 | 1 | """Tests for pipelines: input, output, tool call.""" |
2 | 2 |
|
| 3 | +import logging |
| 4 | + |
| 5 | +import pytest |
| 6 | + |
3 | 7 | from unplug.core.context import ExecutionContext, ToolCall |
4 | 8 | from unplug.core.privacy.secrets import SecretsRegistry, SecretsSanitizer |
5 | 9 | from unplug.core.taint import TaintedText, TrustLevel |
| 10 | +from unplug.guard import Guard |
6 | 11 | from unplug.models import Action, Source |
7 | 12 | from unplug.pipelines.input import InputPipeline |
8 | 13 | from unplug.pipelines.output import OutputPipeline |
@@ -226,3 +231,23 @@ def test_drop_table_in_json_value(self): |
226 | 231 | result = pipeline.run(tc) |
227 | 232 | assert not result.safe |
228 | 233 | assert any(f.category == "destructive" for f in result.findings) |
| 234 | + |
| 235 | + |
| 236 | +class TestPipelineLogging: |
| 237 | + def test_review_outcome_logs_at_info( |
| 238 | + self, |
| 239 | + caplog: pytest.LogCaptureFixture, |
| 240 | + ) -> None: |
| 241 | + guard = Guard() |
| 242 | + guard.scan("Poisoned doc", source=Source.RETRIEVED) |
| 243 | + with caplog.at_level(logging.DEBUG, logger="unplug.pipelines"): |
| 244 | + result = guard.check_tool_call("shell", {"command": "echo hello"}) |
| 245 | + assert result.action == Action.REVIEW |
| 246 | + review_records = [ |
| 247 | + record |
| 248 | + for record in caplog.records |
| 249 | + if record.name == "unplug.pipelines" and "action=review" in record.getMessage() |
| 250 | + ] |
| 251 | + assert review_records |
| 252 | + assert all(record.levelno == logging.INFO for record in review_records) |
| 253 | + assert not any(record.levelno == logging.WARNING for record in review_records) |
0 commit comments