|
21 | 21 | import logging |
22 | 22 | import unittest |
23 | 23 |
|
| 24 | +from socketdev import socketdev |
24 | 25 | from socketdev.fullscans import ( |
25 | | - FullScans, |
26 | 26 | FullScanStreamResponse, |
27 | 27 | SocketArtifact, |
28 | 28 | SocketPURL, |
29 | 29 | SocketPURL_Type, |
30 | 30 | ) |
| 31 | +from tests.support.socket_api_fixture import SocketApiFixture, SocketResponse |
31 | 32 |
|
32 | 33 |
|
33 | 34 | def _artifact_payload(artifact_id: str, purl_type: str) -> dict: |
@@ -109,32 +110,39 @@ def test_malformed_artifact_is_skipped_not_fatal(self): |
109 | 110 | ) |
110 | 111 |
|
111 | 112 | def test_full_scans_stream_skips_artifact_without_id(self): |
112 | | - class Response: |
113 | | - status_code = 200 |
114 | | - text = "\n".join( |
| 113 | + fixture = SocketApiFixture() |
| 114 | + fixture.start() |
| 115 | + try: |
| 116 | + body = "\n".join( |
115 | 117 | json.dumps(artifact) |
116 | 118 | for artifact in ( |
117 | 119 | _artifact_payload("good", "npm"), |
118 | 120 | {"type": "npm", "name": "bad", "alerts": []}, |
119 | 121 | ) |
| 122 | + ).encode("utf-8") |
| 123 | + fixture.route( |
| 124 | + "GET", |
| 125 | + "orgs/org/full-scans/scan", |
| 126 | + SocketResponse(200, None, raw_body=body), |
120 | 127 | ) |
121 | | - |
122 | | - class API: |
123 | | - def do_request(self, **kwargs): |
124 | | - return Response() |
125 | | - |
126 | | - with self.assertLogs("socketdev", level=logging.WARNING) as captured: |
127 | | - response = FullScans(API()).stream("org", "scan", use_types=True) |
128 | | - |
129 | | - self.assertTrue(response.success) |
130 | | - self.assertEqual(list(response.artifacts), ["good"]) |
131 | | - self.assertTrue( |
132 | | - any( |
133 | | - "Skipping artifact without a usable id" in message |
134 | | - for message in captured.output |
135 | | - ), |
136 | | - f"expected a warning about the skipped artifact, got: {captured.output}", |
137 | | - ) |
| 128 | + sdk = socketdev(token="synthetic-token", timeout=1) |
| 129 | + sdk.api.api_url = fixture.base_url |
| 130 | + |
| 131 | + with self.assertLogs("socketdev", level=logging.WARNING) as captured: |
| 132 | + response = sdk.fullscans.stream("org", "scan", use_types=True) |
| 133 | + |
| 134 | + self.assertTrue(response.success) |
| 135 | + self.assertEqual(list(response.artifacts), ["good"]) |
| 136 | + self.assertTrue( |
| 137 | + any( |
| 138 | + "Skipping artifact without a usable id" in message |
| 139 | + for message in captured.output |
| 140 | + ), |
| 141 | + f"expected a warning about the skipped artifact, got: {captured.output}", |
| 142 | + ) |
| 143 | + self.assertEqual(fixture.requests[0].query, {"cached": ["true"]}) |
| 144 | + finally: |
| 145 | + fixture.close() |
138 | 146 |
|
139 | 147 |
|
140 | 148 | if __name__ == "__main__": |
|
0 commit comments