From 230fbe46137354c38c71f677700f396e096e79fe Mon Sep 17 00:00:00 2001 From: RAJVEER42 Date: Sun, 31 May 2026 22:02:12 +0530 Subject: [PATCH] feat(license): wrap POST /license/import-csv endpoint Add `LicenseEndpoint.import_licenses_csv()` to import a CSV file of license rows via the Fossology REST API. Multipart form-data with required `file_input` plus optional `delimiter` and `enclosure`. Returns the server's multi-line summary message (inserted / already-existing rows). Tests cover the live happy path (round-trip a small CSV against the container; the server is idempotent for duplicate shortnames) and a mocked 400 error path. Refs #52 Signed-off-by: RAJVEER42 --- fossology/license.py | 37 +++++++++++++++++++++++++++++ tests/test_license.py | 54 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+) diff --git a/fossology/license.py b/fossology/license.py index 24b6458..0f5b5c1 100644 --- a/fossology/license.py +++ b/fossology/license.py @@ -158,6 +158,43 @@ def add_license(self, license: License, merge_request: bool = False): description = f"Error while adding new license {license.shortName}" raise FossologyApiError(description, response) + def import_licenses_csv( + self, + csv_file: str, + delimiter: str = ",", + enclosure: str = '"', + ) -> str: + """Import licenses from a CSV file. + + API Endpoint: POST /license/import-csv + + :param csv_file: local path to the CSV file containing license data + :param delimiter: field delimiter used in the CSV file (default: ",") + :param enclosure: field enclosure character used in the CSV file (default: '"') + :type csv_file: str + :type delimiter: str + :type enclosure: str + :return: the API response message (multi-line summary of inserted / + already-existing / skipped rows) + :rtype: str + :raises FossologyApiError: if the REST call failed + """ + data = {"delimiter": delimiter, "enclosure": enclosure} + with open(csv_file, "rb") as fp: + response = self.session.post( + f"{self.api}/license/import-csv", + data=data, + files={"file_input": fp}, + ) + + if response.status_code == 200: + message = response.json()["message"] + logger.info(f"License CSV {csv_file} imported: {message}") + return message + + description = f"Unable to import licenses from {csv_file}" + raise FossologyApiError(description, response) + def update_license( self, shortname: str, diff --git a/tests/test_license.py b/tests/test_license.py index c74b96a..7e1c49a 100644 --- a/tests/test_license.py +++ b/tests/test_license.py @@ -32,6 +32,60 @@ def test_another_license(): return License(shortname, fullname, text, url, risk, False) +def test_import_licenses_csv(foss: fossology.Fossology, tmp_path): + csv_path = tmp_path / "licenses.csv" + csv_path.write_text( + "shortname,fullname,text,parent,report,url,risk,group,notes\n" + "CsvImportTestLic,Csv Import Test Lic,License text body,,white," + "http://example.com/csv,5,,\n" + ) + message = foss.import_licenses_csv(str(csv_path)) + # Server returns multi-line summary; either freshly inserted or already + # present from a prior run — both are valid outcomes for an idempotent import. + assert "CsvImportTestLic" in message + assert "Read csv: 1 licenses" in message + + +@responses.activate +def test_import_licenses_csv_error( + foss_server: str, foss: fossology.Fossology, tmp_path +): + responses.add( + responses.POST, + f"{foss_server}/api/v1/license/import-csv", + status=400, + ) + csv_path = tmp_path / "bad.csv" + csv_path.write_text("not really a csv\n") + with pytest.raises(FossologyApiError) as excinfo: + foss.import_licenses_csv(str(csv_path)) + assert f"Unable to import licenses from {csv_path}" in str(excinfo.value) + + +@responses.activate +def test_import_licenses_csv_request_payload( + foss_server: str, foss: fossology.Fossology, tmp_path +): + responses.add( + responses.POST, + f"{foss_server}/api/v1/license/import-csv", + status=200, + json={"code": 200, "message": "head okay\nRead csv: 1 licenses", "type": "INFO"}, + ) + csv_path = tmp_path / "lic.csv" + csv_path.write_text("shortname,fullname\nFoo,Foo License\n") + + message = foss.import_licenses_csv(str(csv_path), delimiter=";", enclosure="'") + + assert "Read csv: 1 licenses" in message + assert len(responses.calls) == 1 + body = responses.calls[0].request.body.decode("utf-8", errors="ignore") + assert 'name="file_input"' in body + assert 'name="delimiter"' in body and "\r\n\r\n;\r\n" in body + assert 'name="enclosure"' in body and "\r\n\r\n'\r\n" in body + assert "shortname,fullname\nFoo,Foo License" in body + + @responses.activate def test_detail_license_error(foss_server: str, foss: fossology.Fossology): responses.add(responses.GET, f"{foss_server}/api/v1/license/Blah", status=500)