Retry, edit, status and header fixes, release 0.3.0

CHANGELOG.md

@@ -5,6 +5,28 @@ All notable changes to `fabric-comanage-api` are documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.3.0] - 2026-10-21
+
+### Changed
+
+- POST requests are no longer retried on 429/5xx responses. POST is not
+  idempotent, so a retry after a server-side success could create a duplicate
+  record. GET, PUT and DELETE are still retried with exponential backoff.
+- `cous_edit`, `coperson_roles_edit` and `ssh_keys_edit` now keep the existing
+  value only when an argument is `None`. Passing `''` clears a field such as a
+  COU description or SSH key comment; previously falsy values were silently
+  ignored. An empty `status`, `affiliation` or `key_type` now raises
+  `ValueError`.
+
+### Fixed
+
+- CoPersonRole `status` is now matched case-insensitively and sent in
+  COmanage's canonical form (`'active'` -> `'Active'`), consistent with
+  affiliation and SSH key type.
+- The HTTP session keeps requests' default headers (`User-Agent`,
+  `Accept-Encoding`, `Accept`) instead of replacing them with `Content-Type`
+  alone.
+
 ## [0.2.3] - 2026-10-21
 
 ### Security

comanage_api/__init__.py

@@ -16,7 +16,7 @@
 from ._sshkeys import SshKeysMixin
 
 # fabric-comanage-api version
-__VERSION__ = "0.2.3"
+__VERSION__ = "0.3.0"
 
 # Library logging: NullHandler prevents "last resort" output for callers
 # who don't configure logging. Callers who want logs should add their own
@@ -99,17 +99,19 @@ def __init__(self, co_api_url: str, co_api_user: str, co_api_pass: str, co_api_o
         self.SSH_KEY_OPTIONS = ['ssh-dss', 'ecdsa-sha2-nistp256', 'ecdsa-sha2-nistp384', 'ecdsa-sha2-nistp521',
                                 'ssh-ed25519', 'ssh-rsa', 'ssh-rsa1']
         # create comanage_api session with retry logic
+        # POST is excluded: it is not idempotent, so retrying a POST that succeeded server-side
+        # but returned an error status could create a duplicate record
         retry = Retry(
             total=3,
             backoff_factor=1,
             status_forcelist=[429, 500, 502, 503, 504],
-            allowed_methods=['GET', 'POST', 'PUT', 'DELETE'],
+            allowed_methods=['GET', 'PUT', 'DELETE'],
         )
         adapter = HTTPAdapter(max_retries=retry)
         self._s = Session()
         self._s.mount('https://', adapter)
         self._s.mount('http://', adapter)
-        self._s.headers = {'Content-Type': 'application/json'}
+        self._s.headers.update({'Content-Type': 'application/json'})
         self._s.auth = (self._CO_API_USER, self._CO_API_PASS)
 
     # HTTP helpers

comanage_api/_copersonroles.py

@@ -8,6 +8,13 @@
 class CoPersonRolesMixin:
     """Mixin providing CoPersonRole API methods."""
 
+    def _normalize_status(self, status: str) -> str:
+        # match case-insensitively, return the canonical COmanage spelling (e.g. 'active' -> 'Active')
+        canonical = {option.lower(): option for option in self.STATUS_OPTIONS}.get(str(status).lower())
+        if canonical is None:
+            raise ValueError("Invalid Fields 'status'")
+        return canonical
+
     def coperson_roles_add(self, coperson_id: int, cou_id: int, status: str = None,
                            affiliation: str = None) -> dict:
         """
@@ -36,9 +43,7 @@ def coperson_roles_add(self, coperson_id: int, cou_id: int, status: str = None,
             'O': str(self._CO_API_ORG_NAME)
         }
         if status:
-            if status not in self.STATUS_OPTIONS:
-                raise ValueError("Invalid Fields 'status'")
-            role['Status'] = str(status)
+            role['Status'] = self._normalize_status(status)
         else:
             role['Status'] = 'Active'
         if affiliation:
@@ -96,18 +101,16 @@ def coperson_roles_edit(self, coperson_role_id: int, coperson_id: int = None, co
             'Version': '1.0',
             'Person': {
                 'Type': 'CO',
-                'Id': str(coperson_id) if coperson_id else str(existing_role.get('Person').get('Id'))
+                'Id': str(coperson_id) if coperson_id is not None else str(existing_role.get('Person').get('Id'))
             },
-            'CouId': str(cou_id) if cou_id else str(existing_role.get('CouId')),
+            'CouId': str(cou_id) if cou_id is not None else str(existing_role.get('CouId')),
             'O': str(self._CO_API_ORG_NAME)
         }
-        if status:
-            if status not in self.STATUS_OPTIONS:
-                raise ValueError("Invalid Fields 'status'")
-            role['Status'] = str(status)
+        if status is not None:
+            role['Status'] = self._normalize_status(status)
         else:
             role['Status'] = existing_role.get('Status')
-        if affiliation:
+        if affiliation is not None:
             affiliation = str(affiliation).lower()
             if affiliation not in self.AFFILIATION_OPTIONS:
                 raise ValueError("Invalid Fields 'affiliation'")

comanage_api/_cous.py

@@ -80,8 +80,8 @@ def cous_edit(self, cou_id: int, name: str = None, description: str = None,
         cou = {
             'Version': '1.0',
             'CoId': self._CO_API_ORG_ID,
-            'Name': str(name) if name else existing_cou.get('Name'),
-            'Description': str(description) if description else existing_cou.get('Description')
+            'Name': str(name) if name is not None else existing_cou.get('Name'),
+            'Description': str(description) if description is not None else existing_cou.get('Description')
         }
         if parent_id is not None and parent_id != 0:
             cou['ParentId'] = str(parent_id)

comanage_api/_sshkeys.py

@@ -107,14 +107,14 @@ def ssh_keys_edit(self, ssh_key_id: int, coperson_id: int = None, ssh_key: str =
             'Version': '1.0',
             'Person': {
                 'Type': 'CO',
-                'Id': str(coperson_id) if coperson_id else str(existing_key.get('Person').get('Id'))
+                'Id': str(coperson_id) if coperson_id is not None else str(existing_key.get('Person').get('Id'))
             },
-            'Skey': str(ssh_key) if ssh_key else existing_key.get('Skey'),
-            'Comment': str(comment) if comment else existing_key.get('Comment'),
-            'SshKeyAuthenticatorId': str(ssh_key_authenticator_id) if ssh_key_authenticator_id
+            'Skey': str(ssh_key) if ssh_key is not None else existing_key.get('Skey'),
+            'Comment': str(comment) if comment is not None else existing_key.get('Comment'),
+            'SshKeyAuthenticatorId': str(ssh_key_authenticator_id) if ssh_key_authenticator_id is not None
             else str(existing_key.get('SshKeyAuthenticatorId'))
         }
-        if key_type:
+        if key_type is not None:
             key_type = str(key_type).lower()
             if key_type not in self.SSH_KEY_OPTIONS:
                 raise ValueError("Invalid Fields 'key_type'")

tests/test_api.py

@@ -51,8 +51,18 @@ def test_retry_adapter_mounted(self, api, mock_adapter):
         adapter = api._s.get_adapter('https://example.com')
         assert isinstance(adapter, HTTPAdapter)
 
+    def test_session_headers_keep_defaults(self, api, mock_adapter):
+        assert api._s.headers['Content-Type'] == 'application/json'
+        assert api._s.headers['User-Agent'].startswith('python-requests/')
+        assert 'Accept-Encoding' in api._s.headers
+
+    def test_retry_excludes_post(self, api, mock_adapter):
+        retry = api._s.get_adapter('https://example.com').max_retries
+        assert 'POST' not in retry.allowed_methods
+        assert {'GET', 'PUT', 'DELETE'} <= set(retry.allowed_methods)
+
     def test_version(self):
-        assert __VERSION__ == '0.2.3'
+        assert __VERSION__ == '0.3.0'
 
 
 class TestOptionSets:

tests/test_copersonroles.py

@@ -47,6 +47,13 @@ def test_custom_status_and_affiliation(self, api, mock_adapter):
         assert body['CoPersonRoles'][0]['Status'] == 'Pending'
         assert body['CoPersonRoles'][0]['Affiliation'] == 'faculty'
 
+    def test_status_case_insensitive(self, api, mock_adapter):
+        mock_adapter.post(f'{API_URL}/co_person_roles.json', json=NEW_ROLE, status_code=201)
+        api.coperson_roles_add(coperson_id=100, cou_id=42, status='active', affiliation='Faculty')
+        body = mock_adapter.last_request.json()
+        assert body['CoPersonRoles'][0]['Status'] == 'Active'
+        assert body['CoPersonRoles'][0]['Affiliation'] == 'faculty'
+
     def test_invalid_status(self, api, mock_adapter):
         with pytest.raises(ValueError, match="status"):
             api.coperson_roles_add(coperson_id=100, cou_id=42, status='BadStatus')
@@ -101,6 +108,18 @@ def test_edit_invalid_affiliation(self, api, mock_adapter):
         with pytest.raises(ValueError, match="affiliation"):
             api.coperson_roles_edit(coperson_role_id=200, affiliation='king')
 
+    def test_edit_empty_status_rejected(self, api, mock_adapter):
+        mock_adapter.get(f'{API_URL}/co_person_roles/200.json', json=SAMPLE_ROLE)
+        with pytest.raises(ValueError, match="status"):
+            api.coperson_roles_edit(coperson_role_id=200, status='')
+
+    def test_edit_status_case_insensitive(self, api, mock_adapter):
+        mock_adapter.get(f'{API_URL}/co_person_roles/200.json', json=SAMPLE_ROLE)
+        mock_adapter.put(f'{API_URL}/co_person_roles/200.json', status_code=200)
+        api.coperson_roles_edit(coperson_role_id=200, status='pendingapproval')
+        body = mock_adapter.last_request.json()
+        assert body['CoPersonRoles'][0]['Status'] == 'PendingApproval'
+
 
 class TestCopersonRolesView:
     def test_view_all(self, api, mock_adapter):

tests/test_cous.py

@@ -84,6 +84,14 @@ def test_edit_keeps_existing_when_none(self, api, mock_adapter):
         assert body['Cous'][0]['Name'] == 'test-cou'
         assert body['Cous'][0]['Description'] == 'A test COU'
 
+    def test_edit_clears_description(self, api, mock_adapter):
+        mock_adapter.get(f'{API_URL}/cous/42.json', json=SAMPLE_COU)
+        mock_adapter.put(f'{API_URL}/cous/42.json', status_code=200)
+        api.cous_edit(cou_id=42, description='')
+        body = mock_adapter.last_request.json()
+        assert body['Cous'][0]['Description'] == ''
+        assert body['Cous'][0]['Name'] == 'test-cou'
+
     def test_parent_id_set(self, api, mock_adapter):
         mock_adapter.get(f'{API_URL}/cous/42.json', json=SAMPLE_COU)
         mock_adapter.put(f'{API_URL}/cous/42.json', status_code=200)

tests/test_sshkeys.py

@@ -96,6 +96,13 @@ def test_edit_keeps_existing(self, api, mock_adapter):
         assert body['SshKeys'][0]['Skey'] == 'AAAAB3NzaC1yc2EAAAADAQABAAABAQC...'
         assert body['SshKeys'][0]['Comment'] == 'test key'
 
+    def test_edit_clears_comment(self, api, mock_adapter):
+        mock_adapter.get(f'{API_URL}/{SSH_PATH}/700.json', json=SAMPLE_SSHKEY)
+        mock_adapter.put(f'{API_URL}/{SSH_PATH}/700.json', status_code=200)
+        api.ssh_keys_edit(ssh_key_id=700, comment='')
+        body = mock_adapter.last_request.json()
+        assert body['SshKeys'][0]['Comment'] == ''
+
     def test_edit_key_type(self, api, mock_adapter):
         mock_adapter.get(f'{API_URL}/{SSH_PATH}/700.json', json=SAMPLE_SSHKEY)
         mock_adapter.put(f'{API_URL}/{SSH_PATH}/700.json', status_code=200)