Merge pull request #169 from GitGuardian/feat/scan-create-incidents-location

Add optional location field to scan_and_create_incidents documents

Author: GG-HH

changelog.d/20260414_172659_hhubert_scan_create_incidents_location.md

@@ -0,0 +1,3 @@
+### Added
+
+- `scan_and_create_incidents`: documents now accept an optional `location` field containing an http/https URL identifying the origin of the document.

pygitguardian/client.py

@@ -545,7 +545,7 @@ def multi_content_scan(
 
     def scan_and_create_incidents(
         self,
-        documents: List[Dict[str, str]],
+        documents: List[Dict[str, Any]],
         source_uuid: UUID,
         *,
         extra_headers: Optional[Dict[str, str]] = None,
@@ -557,8 +557,8 @@ def scan_and_create_incidents(
         character.
 
         :param documents: List of dictionaries containing the keys document
-        and, optionally, filename.
-            example: [{"document":"example content","filename":"intro.py"}]
+        and, optionally, filename and location.
+            example: [{"document":"example content","filename":"intro.py","location":{"url":"https://example.com"}}]
         :param source_uuid: the source UUID that will be used to identify the custom source, for which
         incidents will be created
         :param extra_headers: additional headers to add to the request
@@ -587,8 +587,9 @@ def scan_and_create_incidents(
             "source_uuid": source_uuid,
             "documents": [
                 {
-                    "filename": document["filename"],
-                    "document": document["document"],
+                    key: document[key]
+                    for key in ("filename", "document", "location")
+                    if document.get(key) is not None
                 }
                 for document in request_obj
             ],

pygitguardian/models.py

@@ -37,9 +37,14 @@
 )
 
 
+class DocumentLocationSchema(BaseSchema):
+    url = fields.Url(required=True, schemes={"http", "https"})
+
+
 class DocumentSchema(BaseSchema):
     filename = fields.String(validate=validate.Length(max=256), allow_none=True)
     document = fields.String(required=True)
+    location = fields.Nested(DocumentLocationSchema, allow_none=True, load_default=None)
 
     @staticmethod
     def validate_size(document: Dict[str, Any], maximum_size: int) -> None:
@@ -84,15 +89,24 @@ class Document(Base):
     Attributes:
         filename (optional,str): filename for filename evaluation
         document (str): text content
+        location (optional,dict): origin of the document, with a required http/https url key
     """
 
     SCHEMA = DocumentSchema()
 
-    def __init__(self, document: str, filename: Optional[str] = None, **kwargs: Any):
+    def __init__(
+        self,
+        document: str,
+        filename: Optional[str] = None,
+        location: Optional[Dict[str, str]] = None,
+        **kwargs: Any,
+    ):
         super().__init__()
         self.document = document
         if filename:
             self.filename = filename
+        if location:
+            self.location = location
 
     def __repr__(self) -> str:
         return f"filename:{self.filename}, document:{self.document}"

tests/test_client.py

@@ -698,6 +698,47 @@ def test_scan_and_create_incidents_payload_structure(client: GGClient):
     assert mock_response.call_count == 1
 
 
+@responses.activate
+def test_scan_and_create_incidents_payload_with_location(client: GGClient):
+    """
+    GIVEN a ggclient
+    WHEN calling scan_and_create_incidents with a document location
+    THEN the location is included in the payload
+    """
+
+    location = {"url": "https://wiki.example.com/my-config-page"}
+    documents = [{"filename": FILENAME, "document": DOCUMENT, "location": location}]
+    source_uuid = "123e4567-e89b-12d3-a456-426614174000"
+
+    expected_payload = {
+        "documents": [
+            {
+                "document": DOCUMENT,
+                "filename": FILENAME,
+                "location": location,
+            }
+        ],
+        "source_uuid": source_uuid,
+    }
+
+    mock_response = responses.post(
+        url=client._url_from_endpoint("scan/create-incidents", "v1"),
+        status=200,
+        match=[matchers.json_params_matcher(expected_payload)],
+        json=[
+            {
+                "policy_break_count": 0,
+                "policies": ["pol"],
+                "policy_breaks": [],
+            }
+        ],
+    )
+
+    client.scan_and_create_incidents(documents, source_uuid)
+
+    assert mock_response.call_count == 1
+
+
 @responses.activate
 def test_retrieve_secret_incident(client: GGClient):
     """