Allow to set a custom user agent

src/peppol_py/as4_sender.py

@@ -29,9 +29,9 @@ def get_headers_and_body_for_posting_as4_document(document_content, document_xml
 
     return body, headers
 
-def post_edelivery_as4_document(endpoint_url, body, headers, timeout):
+def post_edelivery_as4_document(endpoint_url, body, headers, timeout, user_agent):
     try:
-        r = requests.post(endpoint_url, data=body, headers=headers, timeout=timeout)
+        r = requests.post(endpoint_url, data=body, headers={"User-Agent": user_agent, **headers}, timeout=timeout)
     except (ConnectionError, requests.exceptions.RequestException) as e:
         raise make_sendpeppol_error(str(e), 'server-error', temporary=True, url=endpoint_url)
 

src/peppol_py/sender.py

@@ -120,6 +120,7 @@ def send_peppol_document(
     test_environment: bool=True,
     timeout: int=20,
     dryrun: bool=False,
+    user_agent: str="peppol-py",
 ) -> dict:
     """
     Send a peppol document. Returned is a dictionary of information you need to record to later send reports to Peppol.
@@ -151,6 +152,8 @@ def send_peppol_document(
 
     ``dryrun`` (bool): if specified, will prepare, get the endpoint, test document for validation errors but not send to remote endpoint.
                        Return value will be a tuple of ``body, header, stats``.
+
+    ``user_agent`` (str): if specified, will replace the user agent sent on HTTP connections
     """
     document_xml = etree.fromstring(document_content)
     #print(etree.tostring(document_xml, pretty_print=True).decode())
@@ -176,7 +179,7 @@ def send_peppol_document(
 
     document_content, document_xml = wrap_ubl_in_peppol_standard_business_document_header(document_xml, utc_timestamp, document_type, process_type, sender_id, sender_country, receiver_id)
 
-    transport_profile, endpoint_url, receiver_cert = get_service_info_for_participant_from_smp(receiver_id, document_type, test_environment=test_environment, timeout=timeout)
+    transport_profile, endpoint_url, receiver_cert = get_service_info_for_participant_from_smp(receiver_id, document_type, test_environment=test_environment, timeout=timeout, user_agent=user_agent)
 
     if not endpoint_url:
         raise make_sendpeppol_error("Endpoint URL not found", 'not-found-in-smp')
@@ -206,6 +209,6 @@ def send_peppol_document(
     if dryrun:
         return body, headers, stats
 
-    post_edelivery_as4_document(endpoint_url, body, headers, timeout=timeout)
+    post_edelivery_as4_document(endpoint_url, body, headers, timeout=timeout, user_agent=user_agent)
 
     return stats

src/peppol_py/smp.py

@@ -50,12 +50,12 @@ def get_smp_url_from_dns(participant_id, test_environment):
     return result.rstrip("/") + "/iso6523-actorid-upis::" + participant_id
 
 
-def get_service_urls_for_participant_from_smp(participant_id, test_environment, timeout):
+def get_service_urls_for_participant_from_smp(participant_id, test_environment, timeout, user_agent):
     # SML: receiver -> SMP domain (DNS)
     smp_url = get_smp_url_from_dns(participant_id, test_environment)
 
     try:
-        r = requests.get(smp_url, timeout=timeout)
+        r = requests.get(smp_url, timeout=timeout, headers={"User-Agent": user_agent})
         r.raise_for_status()
     except (ConnectionError, requests.exceptions.RequestException) as e:
         temporary = True
@@ -80,18 +80,18 @@ def get_service_urls_for_participant_from_smp(participant_id, test_environment,
     return service_urls
 
 
-def get_service_info_for_participant_from_smp(participant_id, document_type, test_environment, timeout):
+def get_service_info_for_participant_from_smp(participant_id, document_type, test_environment, timeout, user_agent):
     """Lookup participant in Peppol Service Metadata Publisher for
     information on service type. Service type is Scope.Identifier +
     '::' + Scope.InstanceIdentifier from the Peppol header."""
 
-    service_urls = get_service_urls_for_participant_from_smp(participant_id, test_environment, timeout)
+    service_urls = get_service_urls_for_participant_from_smp(participant_id, test_environment, timeout, user_agent)
     service_url = next((url for url in service_urls if url.endswith(urllib.parse.quote(document_type))), None)
     if not service_url:
         raise make_sendpeppol_error("{0} not found in {1}".format(document_type, [urllib.parse.unquote(url) for url in service_urls]), 'not-found-in-smp')
 
     try:
-        r = requests.get(service_url, timeout=timeout)
+        r = requests.get(service_url, timeout=timeout, headers={"User-Agent": user_agent})
         r.raise_for_status()
     except (ConnectionError, requests.exceptions.RequestException) as e:
         temporary = True