From: Shweta Sodani Date: Tue, 26 May 2026 14:27:42 +0000 (+0530) Subject: mgr/smb: Add test coverage for RGW share functionality X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=9a80169005e2f8401b979f03dcaf07f7db903792;p=ceph.git mgr/smb: Add test coverage for RGW share functionality Add test cases for RGW-backed SMB shares Key changes: - RGWStorage creation, validation, and serialization - Edge cases: None values, mutual exclusion Signed-off-by: Shweta Sodani Signed-off-by: Avan Thakkar --- diff --git a/src/pybind/mgr/cephadm/tests/services/test_smb.py b/src/pybind/mgr/cephadm/tests/services/test_smb.py index 2add9e5fe6e..6c942096392 100644 --- a/src/pybind/mgr/cephadm/tests/services/test_smb.py +++ b/src/pybind/mgr/cephadm/tests/services/test_smb.py @@ -55,6 +55,7 @@ class TestSMB: 'cluster_id': 'foxtrot', 'features': [], 'config_uri': 'rados://.smb/foxtrot/config.json', + 'extra_config_uris': ['rados:mon-config-key:smb/config/foxtrot/config.smb.rgw'], 'config': '', 'keyring': '[client.smb.config.tango.briskly]\nkey = None\n', 'config_auth_entity': 'client.smb.config.tango.briskly', @@ -125,6 +126,7 @@ class TestSMB: 'cluster_id': 'foxtrot', 'features': ['domain'], 'config_uri': 'rados://.smb/foxtrot/config2.json', + 'extra_config_uris': ['rados:mon-config-key:smb/config/foxtrot/config.smb.rgw'], 'join_sources': [ 'rados://.smb/foxtrot/join1.json', 'rados:mon-config-key:smb/config/foxtrot/join2.json', diff --git a/src/pybind/mgr/smb/tests/test_handler.py b/src/pybind/mgr/smb/tests/test_handler.py index 868958999f2..eb470cdc414 100644 --- a/src/pybind/mgr/smb/tests/test_handler.py +++ b/src/pybind/mgr/smb/tests/test_handler.py @@ -4,6 +4,14 @@ import smb from smb.handler import _FakeEarmarkResolver +class _FakeToolExecer: + """Mock tool executor for testing.""" + + def tool_exec(self, cmd: list[str]) -> tuple[int, str, str]: + """Mock tool_exec that returns success.""" + return (0, '{}', '') + + def _cluster(**kwargs): if 'clustering' not in kwargs: kwargs['clustering'] = smb.enums.SMBClustering.NEVER @@ -25,6 +33,8 @@ def thandler(): # into a single store. Do that to simplify testing a bit. public_store=ext_store, priv_store=ext_store, + mon_cmd_issuer=None, + tool_execer=_FakeToolExecer(), ) diff --git a/src/pybind/mgr/smb/tests/test_resources.py b/src/pybind/mgr/smb/tests/test_resources.py index a9987b2a7ce..dcbca7f0ac6 100644 --- a/src/pybind/mgr/smb/tests/test_resources.py +++ b/src/pybind/mgr/smb/tests/test_resources.py @@ -1283,3 +1283,308 @@ def test_share_qos_remove_individual_limit(): assert updated_cephfs.qos.write_iops_limit == 200 # Preserved assert updated_cephfs.qos.read_bw_limit == "10M" # Preserved assert updated_cephfs.qos.write_bw_limit == "20M" # Preserved + + +def test_rgw_storage_basic(): + """Test basic RGWStorage creation and validation.""" + import yaml + + yaml_str = """ +resource_type: ceph.smb.share +cluster_id: rgwcluster +share_id: rgwshare +name: RGW Share +rgw: + bucket: my-bucket +""" + data = yaml.safe_load_all(yaml_str) + loaded = smb.resources.load(data) + assert loaded + + share = loaded[0] + assert isinstance(share.rgw, smb.resources.RGWStorage) + assert share.rgw.bucket == 'my-bucket' + assert share.rgw.user_id is None + assert share.rgw.credential_ref is None + + +def test_rgw_storage_with_credentials(): + """Test RGWStorage with credential_ref.""" + import yaml + + yaml_str = """ +resource_type: ceph.smb.share +cluster_id: rgwcluster +share_id: rgwshare +name: RGW Share +rgw: + bucket: my-bucket + user_id: testuser + credential_ref: testuser +""" + data = yaml.safe_load_all(yaml_str) + loaded = smb.resources.load(data) + assert loaded + + share = loaded[0] + assert share.rgw.bucket == 'my-bucket' + assert share.rgw.user_id == 'testuser' + assert share.rgw.credential_ref == 'testuser' + + +def test_rgw_storage_missing_bucket(): + """Test validation error when bucket is missing.""" + import yaml + + yaml_str = """ +resource_type: ceph.smb.share +cluster_id: rgwcluster +share_id: rgwshare +name: RGW Share +rgw: + bucket: "" +""" + data = yaml.safe_load_all(yaml_str) + with pytest.raises(ValueError, match='bucket requires a value'): + smb.resources.load(data) + + +def test_rgw_storage_convert_mask(): + """Test RGWStorage conversion - credentials now in RGWCredential.""" + storage = smb.resources.RGWStorage( + bucket='my-bucket', + user_id='testuser', + credential_ref='testuser', + ) + + masked = storage.convert( + (smb.enums.PasswordFilter.NONE, smb.enums.PasswordFilter.HIDDEN) + ) + assert masked.bucket == 'my-bucket' + assert masked.user_id == 'testuser' + assert masked.credential_ref == 'testuser' + + +def test_rgw_storage_convert_none_credentials(): + """Test conversion when credential_ref is None.""" + storage = smb.resources.RGWStorage( + bucket='my-bucket', + ) + + encoded = storage.convert( + (smb.enums.PasswordFilter.NONE, smb.enums.PasswordFilter.BASE64) + ) + assert encoded.credential_ref is None + + masked = storage.convert( + (smb.enums.PasswordFilter.NONE, smb.enums.PasswordFilter.HIDDEN) + ) + assert masked.credential_ref is None + + +def test_rgw_storage_to_simplified(): + """Test RGWStorage serialization to simplified format.""" + import yaml + + yaml_str = """ +resource_type: ceph.smb.share +cluster_id: rgwcluster +share_id: rgwshare +name: RGW Share +rgw: + bucket: my-bucket + user_id: testuser + credential_ref: testuser +""" + data = yaml.safe_load_all(yaml_str) + loaded = smb.resources.load(data) + assert loaded + + share = loaded[0] + simplified = share.to_simplified() + + assert 'rgw' in simplified + assert simplified['rgw']['bucket'] == 'my-bucket' + assert simplified['rgw']['user_id'] == 'testuser' + # Credentials are now referenced via credential_ref + assert simplified['rgw']['credential_ref'] == 'testuser' + + +def test_rgw_storage_invalid_credentials(): + """Test RGWStorage with invalid credential_ref.""" + import yaml + + yaml_str = """ +resource_type: ceph.smb.share +cluster_id: rgwcluster +share_id: rgwshare +name: RGW Share +rgw: + bucket: my-bucket + credential_ref: "invalid-ref-with-spaces" +""" + data = yaml.safe_load_all(yaml_str) + # Note: Credential validation happens during staging, not at load time + # This test documents that RGWStorage accepts any credential_ref string + loaded = smb.resources.load(data) + assert loaded + assert loaded[0].rgw.credential_ref == "invalid-ref-with-spaces" + + +def test_share_with_rgw_and_cephfs_mutual_exclusion(): + """Test that share cannot have both rgw and cephfs.""" + import yaml + + yaml_str = """ +resource_type: ceph.smb.share +cluster_id: testcluster +share_id: testshare +name: Invalid Share +cephfs: + volume: cephfs + path: / +rgw: + bucket: my-bucket + path: / +""" + data = yaml.safe_load_all(yaml_str) + with pytest.raises(ValueError, match='only one storage backend'): + smb.resources.load(data) + + +def test_rgw_credential_basic(): + """Test basic RGWCredential creation and validation.""" + import yaml + + yaml_str = """ +resource_type: ceph.smb.rgw.credential +rgw_credential_id: rgwcred1 +user_id: s3user +access_key_id: AKIAIOSFODNN7EXAMPLE +secret_access_key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY +""" + data = yaml.safe_load_all(yaml_str) + loaded = smb.resources.load(data) + assert loaded + + cred = loaded[0] + assert isinstance(cred, smb.resources.RGWCredential) + assert cred.rgw_credential_id == 'rgwcred1' + assert cred.user_id == 's3user' + assert cred.access_key_id == 'AKIAIOSFODNN7EXAMPLE' + assert ( + cred.secret_access_key == 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY' + ) + assert cred.intent == smb.enums.Intent.PRESENT + + +def test_rgw_credential_missing_required_fields(): + """Test that RGWCredential requires user_id, access_key_id, and secret_access_key.""" + import yaml + + # Missing user_id + yaml_str = """ +resource_type: ceph.smb.rgw.credential +rgw_credential_id: rgwcred1 +access_key_id: AKIAIOSFODNN7EXAMPLE +secret_access_key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY +""" + data = yaml.safe_load_all(yaml_str) + with pytest.raises(Exception): # Will fail during construction + smb.resources.load(data) + + # Missing access_key_id + yaml_str = """ +resource_type: ceph.smb.rgw.credential +rgw_credential_id: rgwcred1 +user_id: s3user +secret_access_key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY +""" + data = yaml.safe_load_all(yaml_str) + with pytest.raises(Exception): # Will fail during construction + smb.resources.load(data) + + # Missing secret_access_key + yaml_str = """ +resource_type: ceph.smb.rgw.credential +rgw_credential_id: rgwcred1 +user_id: s3user +access_key_id: AKIAIOSFODNN7EXAMPLE +""" + data = yaml.safe_load_all(yaml_str) + with pytest.raises(Exception): # Will fail during construction + smb.resources.load(data) + + +def test_rgw_credential_password_conversion(): + """Test RGWCredential password field conversion.""" + cred = smb.resources.RGWCredential( + rgw_credential_id='test_cred', + user_id='testuser', + access_key_id='AKIATEST', + secret_access_key='secretkey123', + ) + + # Test masking (NONE -> HIDDEN) + masked = cred.convert( + (smb.enums.PasswordFilter.NONE, smb.enums.PasswordFilter.HIDDEN) + ) + assert masked.access_key_id == '****************' + assert masked.secret_access_key == '****************' + assert masked.user_id == 'testuser' # user_id should not be masked + + # Test base64 encoding (NONE -> BASE64) + encoded = cred.convert( + (smb.enums.PasswordFilter.NONE, smb.enums.PasswordFilter.BASE64) + ) + assert encoded.access_key_id != 'AKIATEST' + assert encoded.secret_access_key != 'secretkey123' + assert encoded.user_id == 'testuser' + + # Test base64 decoding (BASE64 -> NONE) + decoded = encoded.convert( + (smb.enums.PasswordFilter.BASE64, smb.enums.PasswordFilter.NONE) + ) + assert decoded.access_key_id == 'AKIATEST' + assert decoded.secret_access_key == 'secretkey123' + + +def test_rgw_credential_with_linked_cluster(): + """Test RGWCredential with linked_to_cluster.""" + import yaml + + yaml_str = """ +resource_type: ceph.smb.rgw.credential +rgw_credential_id: rgwcred1 +user_id: s3user +access_key_id: AKIAIOSFODNN7EXAMPLE +secret_access_key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY +linked_to_cluster: mysmb +""" + data = yaml.safe_load_all(yaml_str) + loaded = smb.resources.load(data) + assert loaded + + cred = loaded[0] + assert cred.linked_to_cluster == 'mysmb' + + +def test_rgw_credential_removed(): + """Test RGWCredential with removed intent.""" + import yaml + + yaml_str = """ +resource_type: ceph.smb.rgw.credential +rgw_credential_id: rgwcred1 +user_id: placeholder +access_key_id: placeholder +secret_access_key: placeholder +intent: removed +""" + data = yaml.safe_load_all(yaml_str) + loaded = smb.resources.load(data) + assert loaded + + cred = loaded[0] + assert cred.intent == smb.enums.Intent.REMOVED + assert cred.rgw_credential_id == 'rgwcred1' diff --git a/src/pybind/mgr/smb/tests/test_rgw_shares.py b/src/pybind/mgr/smb/tests/test_rgw_shares.py new file mode 100644 index 00000000000..2983e81c57d --- /dev/null +++ b/src/pybind/mgr/smb/tests/test_rgw_shares.py @@ -0,0 +1,508 @@ +import json + +import pytest + +import smb +import smb.external + + +class _FakeToolExecer: + """Mock tool executor for testing RGW operations.""" + + def tool_exec(self, cmd: list[str]) -> tuple[int, str, str]: + """Mock tool_exec that returns success for RGW commands.""" + # Check if this is a radosgw-admin bucket stats command + if 'radosgw-admin' in cmd and 'bucket' in cmd and 'stats' in cmd: + # Return mock bucket stats with owner field + bucket_stats = json.dumps( + {'owner': 'testuser', 'bucket': 'my-bucket', 'usage': {}} + ) + return (0, bucket_stats, '') + # Check if this is a radosgw-admin user info command + if 'radosgw-admin' in cmd and 'user' in cmd and 'info' in cmd: + # Return mock user info with credentials + user_info = json.dumps( + { + 'user_id': 'testuser', + 'keys': [ + { + 'access_key': 'AUTO_FETCHED_ACCESS_KEY', + 'secret_key': 'AUTO_FETCHED_SECRET_KEY', + } + ], + } + ) + return (0, user_info, '') + # Default response for other commands + return (0, '{}', '') + + +def _cluster(**kwargs): + """Helper to create cluster with default clustering setting.""" + if 'clustering' not in kwargs: + kwargs['clustering'] = smb.enums.SMBClustering.NEVER + return smb.resources.Cluster(**kwargs) + + +@pytest.fixture +def thandler(): + """Fixture for RGW tests with RGW-aware tool executor.""" + ext_store = smb.config_store.MemConfigStore() + return smb.handler.ClusterConfigHandler( + internal_store=smb.config_store.MemConfigStore(), + public_store=ext_store, + priv_store=ext_store, + mon_cmd_issuer=None, + tool_execer=_FakeToolExecer(), + ) + + +def test_internal_apply_cluster_and_rgw_share(thandler): + """Test creating a cluster with an RGW share.""" + cluster = _cluster( + cluster_id='rgwtest', + auth_mode=smb.enums.AuthMode.USER, + user_group_settings=[ + smb.resources.UserGroupSource( + source_type=smb.resources.UserGroupSourceType.EMPTY, + ), + ], + ) + share = smb.resources.Share( + cluster_id='rgwtest', + share_id='rgwshare1', + name='RGW Share One', + rgw=smb.resources.RGWStorage( + bucket='my-bucket', + user_id='testuser', + ), + ) + rg = thandler.apply([cluster, share]) + assert rg.success, rg.to_simplified() + assert ('clusters', 'rgwtest') in thandler.internal_store.data + assert ('shares', 'rgwtest.rgwshare1') in thandler.internal_store.data + # Check that credential was auto-created + assert ('rgw_creds', 'testuser') in thandler.internal_store.data + + shares = thandler.share_ids() + assert len(shares) == 1 + assert ('rgwtest', 'rgwshare1') in shares + + # Verify share uses credential_ref + share_data = thandler.internal_store.data[('shares', 'rgwtest.rgwshare1')] + assert share_data['rgw']['credential_ref'] == 'testuser' + + +def test_apply_rgw_share_with_credentials(thandler): + """Test applying an RGW share with full credentials.""" + cluster = _cluster( + cluster_id='rgwcluster', + auth_mode=smb.enums.AuthMode.USER, + user_group_settings=[ + smb.resources.UserGroupSource( + source_type=smb.resources.UserGroupSourceType.EMPTY, + ), + ], + ) + share = smb.resources.Share( + cluster_id='rgwcluster', + share_id='s3share', + name='S3 Share', + rgw=smb.resources.RGWStorage( + bucket='test-bucket', + user_id='rgwuser', + ), + ) + rg = thandler.apply([cluster, share]) + assert rg.success, rg.to_simplified() + + # Verify credential was auto-created with fetched credentials + assert ('rgw_creds', 'rgwuser') in thandler.internal_store.data + cred_dict = thandler.internal_store.data[('rgw_creds', 'rgwuser')] + assert cred_dict['user_id'] == 'rgwuser' + assert cred_dict['access_key_id'] == 'AUTO_FETCHED_ACCESS_KEY' + assert cred_dict['secret_access_key'] == 'AUTO_FETCHED_SECRET_KEY' + + # Verify share uses credential_ref + share_dict = thandler.internal_store.data[ + ('shares', 'rgwcluster.s3share') + ] + assert share_dict['rgw']['bucket'] == 'test-bucket' + assert share_dict['rgw']['credential_ref'] == 'rgwuser' + + +def test_rgw_share_auto_fetch_credentials(thandler): + """Test RGW share with auto-fetched credentials (bucket only provided).""" + cluster = _cluster( + cluster_id='autofetch', + auth_mode=smb.enums.AuthMode.USER, + user_group_settings=[ + smb.resources.UserGroupSource( + source_type=smb.resources.UserGroupSourceType.EMPTY, + ), + ], + ) + share = smb.resources.Share( + cluster_id='autofetch', + share_id='autoshare', + name='Auto Fetch Share', + rgw=smb.resources.RGWStorage( + bucket='my-bucket', + ), + ) + rg = thandler.apply([cluster, share]) + assert rg.success, rg.to_simplified() + + # Verify credential was auto-created with fetched credentials + assert ('rgw_creds', 'testuser') in thandler.internal_store.data + cred_dict = thandler.internal_store.data[('rgw_creds', 'testuser')] + assert cred_dict['user_id'] == 'testuser' + assert cred_dict['access_key_id'] == 'AUTO_FETCHED_ACCESS_KEY' + assert cred_dict['secret_access_key'] == 'AUTO_FETCHED_SECRET_KEY' + + # Verify share uses credential_ref + share_dict = thandler.internal_store.data[ + ('shares', 'autofetch.autoshare') + ] + assert share_dict['rgw']['bucket'] == 'my-bucket' + assert share_dict['rgw']['credential_ref'] == 'testuser' + + +def test_rgw_share_remove(thandler): + """Test removing an RGW share.""" + # First create a cluster and RGW share + cluster = _cluster( + cluster_id='rgwtest', + auth_mode=smb.enums.AuthMode.USER, + user_group_settings=[ + smb.resources.UserGroupSource( + source_type=smb.resources.UserGroupSourceType.EMPTY, + ), + ], + ) + share = smb.resources.Share( + cluster_id='rgwtest', + share_id='rgwshare1', + name='RGW Share One', + rgw=smb.resources.RGWStorage( + bucket='my-bucket', + user_id='testuser', + ), + ) + rg = thandler.apply([cluster, share]) + assert rg.success, rg.to_simplified() + + # Now remove the share + rmshare = smb.resources.RemovedShare( + cluster_id='rgwtest', + share_id='rgwshare1', + ) + rg = thandler.apply([rmshare]) + assert rg.success, rg.to_simplified() + + shares = thandler.share_ids() + assert len(shares) == 0 + assert ('shares', 'rgwtest.rgwshare1') not in thandler.internal_store.data + + +def test_rgw_share_minimal_config(thandler): + """Test RGW share with minimal configuration (bucket only).""" + cluster = _cluster( + cluster_id='minimalrgw', + auth_mode=smb.enums.AuthMode.USER, + user_group_settings=[ + smb.resources.UserGroupSource( + source_type=smb.resources.UserGroupSourceType.EMPTY, + ), + ], + ) + share = smb.resources.Share( + cluster_id='minimalrgw', + share_id='minimal', + name='Minimal RGW Share', + rgw=smb.resources.RGWStorage( + bucket='my-bucket', + ), + ) + rg = thandler.apply([cluster, share]) + assert rg.success, rg.to_simplified() + + # Verify credential was auto-created with fetched credentials + assert ('rgw_creds', 'testuser') in thandler.internal_store.data + cred_dict = thandler.internal_store.data[('rgw_creds', 'testuser')] + assert cred_dict['user_id'] == 'testuser' + assert cred_dict['access_key_id'] == 'AUTO_FETCHED_ACCESS_KEY' + assert cred_dict['secret_access_key'] == 'AUTO_FETCHED_SECRET_KEY' + + # Verify share uses credential_ref + share_dict = thandler.internal_store.data[ + ('shares', 'minimalrgw.minimal') + ] + assert share_dict['rgw']['bucket'] == 'my-bucket' + assert share_dict['rgw']['credential_ref'] == 'testuser' + + +def test_multiple_rgw_shares_same_cluster(thandler): + """Test creating multiple RGW shares in the same cluster.""" + cluster = _cluster( + cluster_id='multirgw', + auth_mode=smb.enums.AuthMode.USER, + user_group_settings=[ + smb.resources.UserGroupSource( + source_type=smb.resources.UserGroupSourceType.EMPTY, + ), + ], + ) + share1 = smb.resources.Share( + cluster_id='multirgw', + share_id='share1', + name='RGW Share 1', + rgw=smb.resources.RGWStorage( + bucket='bucket1', + user_id='user1', + ), + ) + share2 = smb.resources.Share( + cluster_id='multirgw', + share_id='share2', + name='RGW Share 2', + rgw=smb.resources.RGWStorage( + bucket='bucket2', + user_id='user2', + ), + ) + rg = thandler.apply([cluster, share1, share2]) + assert rg.success, rg.to_simplified() + + shares = thandler.share_ids() + assert len(shares) == 2 + assert ('multirgw', 'share1') in shares + assert ('multirgw', 'share2') in shares + + # Verify credentials were auto-created for both users + assert ('rgw_creds', 'user1') in thandler.internal_store.data + assert ('rgw_creds', 'user2') in thandler.internal_store.data + + # Verify both shares use credential_ref + share1_dict = thandler.internal_store.data[('shares', 'multirgw.share1')] + share2_dict = thandler.internal_store.data[('shares', 'multirgw.share2')] + assert share1_dict['rgw']['bucket'] == 'bucket1' + assert share1_dict['rgw']['credential_ref'] == 'user1' + assert share2_dict['rgw']['bucket'] == 'bucket2' + assert share2_dict['rgw']['credential_ref'] == 'user2' + + +def test_rgw_credential_reuse_same_user(thandler): + """Test that multiple shares with same user_id reuse the same credential.""" + cluster = _cluster( + cluster_id='reusetest', + auth_mode=smb.enums.AuthMode.USER, + user_group_settings=[ + smb.resources.UserGroupSource( + source_type=smb.resources.UserGroupSourceType.EMPTY, + ), + ], + ) + share1 = smb.resources.Share( + cluster_id='reusetest', + share_id='share1', + name='RGW Share 1', + rgw=smb.resources.RGWStorage( + bucket='bucket1', + user_id='shareduser', + ), + ) + share2 = smb.resources.Share( + cluster_id='reusetest', + share_id='share2', + name='RGW Share 2', + rgw=smb.resources.RGWStorage( + bucket='bucket2', + user_id='shareduser', + ), + ) + rg = thandler.apply([cluster, share1, share2]) + assert rg.success, rg.to_simplified() + + # Verify only ONE credential was created for the shared user + rgw_creds = [ + k for k in thandler.internal_store.data.keys() if k[0] == 'rgw_creds' + ] + assert len(rgw_creds) == 1 + assert ('rgw_creds', 'shareduser') in thandler.internal_store.data + + # Verify both shares reference the same credential + share1_dict = thandler.internal_store.data[('shares', 'reusetest.share1')] + share2_dict = thandler.internal_store.data[('shares', 'reusetest.share2')] + assert share1_dict['rgw']['credential_ref'] == 'shareduser' + assert share2_dict['rgw']['credential_ref'] == 'shareduser' + + +# ---- priv-store credential isolation tests ---- +# These tests use SEPARATE public and private stores so each can be +# independently inspected. The real credential values must +# only ever appear in the private store (via a config:merge stub), never in +# the public RADOS config. + + +@pytest.fixture +def split_handler(): + """Handler with separate public and private stores.""" + pub = smb.config_store.MemConfigStore() + priv = smb.config_store.MemConfigStore() + h = smb.handler.ClusterConfigHandler( + internal_store=smb.config_store.MemConfigStore(), + public_store=pub, + priv_store=priv, + tool_execer=_FakeToolExecer(), + ) + return h, pub, priv + + +def _rgw_cluster_and_share(cluster_id, share_id, share_name, **rgw_kwargs): + cluster = _cluster( + cluster_id=cluster_id, + auth_mode=smb.enums.AuthMode.USER, + user_group_settings=[ + smb.resources.UserGroupSource( + source_type=smb.resources.UserGroupSourceType.EMPTY, + ), + ], + ) + share = smb.resources.Share( + cluster_id=cluster_id, + share_id=share_id, + name=share_name, + rgw=smb.resources.RGWStorage(**rgw_kwargs), + ) + return cluster, share + + +def test_rgw_credentials_absent_from_public_config(split_handler): + """Credential values must be empty strings in the public RADOS config.""" + h, pub, priv = split_handler + cluster, share = _rgw_cluster_and_share( + 'c1', + 's1', + 'mybucket', + bucket='mybucket', + user_id='testuser', + ) + rg = h.apply([cluster, share]) + assert rg.success, rg.to_simplified() + + # Verify credential was auto-created + assert ('rgw_creds', 'testuser') in h.internal_store.data + + pub_cfg = pub['c1', 'config.smb'].get() + opts = pub_cfg['shares']['mybucket']['options'] + assert opts['ceph_rgw:access_key'] == '' + assert opts['ceph_rgw:secret_access_key'] == '' + + +def test_rgw_credential_stub_written_to_priv_store(split_handler): + """Private store must hold config:merge stub with real credential values.""" + h, pub, priv = split_handler + cluster, share = _rgw_cluster_and_share( + 'c1', + 's1', + 'mybucket', + bucket='mybucket', + user_id='testuser', + ) + rg = h.apply([cluster, share]) + assert rg.success, rg.to_simplified() + + # Verify credential was auto-created with fetched credentials + assert ('rgw_creds', 'testuser') in h.internal_store.data + cred_dict = h.internal_store.data[('rgw_creds', 'testuser')] + assert cred_dict['access_key_id'] == 'AUTO_FETCHED_ACCESS_KEY' + assert cred_dict['secret_access_key'] == 'AUTO_FETCHED_SECRET_KEY' + + stub = priv['c1', 'config.smb.rgw'].get() + assert stub['samba-container-config'] == 'v0' + assert 'config:merge' in stub + opts = stub['config:merge']['shares']['mybucket']['options'] + assert opts['ceph_rgw:access_key'] == 'AUTO_FETCHED_ACCESS_KEY' + assert opts['ceph_rgw:secret_access_key'] == 'AUTO_FETCHED_SECRET_KEY' + + +def test_no_priv_store_entry_for_non_rgw_cluster(split_handler): + """A cluster with no RGW shares must not write a credential stub.""" + h, pub, priv = split_handler + cluster = _cluster( + cluster_id='cephfs1', + auth_mode=smb.enums.AuthMode.USER, + user_group_settings=[ + smb.resources.UserGroupSource( + source_type=smb.resources.UserGroupSourceType.EMPTY, + ), + ], + ) + share = smb.resources.Share( + cluster_id='cephfs1', + share_id='fsshare', + name='FS Share', + cephfs=smb.resources.CephFSStorage( + volume='cephfs', + path='/', + ), + ) + rg = h.apply([cluster, share]) + assert rg.success, rg.to_simplified() + + stub_entry = priv['cephfs1', 'config.smb.rgw'] + assert not stub_entry.exists() + + +def test_multi_share_stub_covers_all_rgw_shares(split_handler): + """Credential stub must contain entries for every RGW share.""" + h, pub, priv = split_handler + cluster = _cluster( + cluster_id='multi', + auth_mode=smb.enums.AuthMode.USER, + user_group_settings=[ + smb.resources.UserGroupSource( + source_type=smb.resources.UserGroupSourceType.EMPTY, + ), + ], + ) + share1 = smb.resources.Share( + cluster_id='multi', + share_id='s1', + name='bucket1', + rgw=smb.resources.RGWStorage( + bucket='bucket1', + user_id='u1', + ), + ) + share2 = smb.resources.Share( + cluster_id='multi', + share_id='s2', + name='bucket2', + rgw=smb.resources.RGWStorage( + bucket='bucket2', + user_id='u2', + ), + ) + rg = h.apply([cluster, share1, share2]) + assert rg.success, rg.to_simplified() + + # Verify credentials were auto-created for both users with fetched credentials + assert ('rgw_creds', 'u1') in h.internal_store.data + assert ('rgw_creds', 'u2') in h.internal_store.data + cred1 = h.internal_store.data[('rgw_creds', 'u1')] + cred2 = h.internal_store.data[('rgw_creds', 'u2')] + assert cred1['access_key_id'] == 'AUTO_FETCHED_ACCESS_KEY' + assert cred1['secret_access_key'] == 'AUTO_FETCHED_SECRET_KEY' + assert cred2['access_key_id'] == 'AUTO_FETCHED_ACCESS_KEY' + assert cred2['secret_access_key'] == 'AUTO_FETCHED_SECRET_KEY' + + stub = priv['multi', 'config.smb.rgw'].get() + merge_shares = stub['config:merge']['shares'] + b1opts = merge_shares['bucket1']['options'] + b2opts = merge_shares['bucket2']['options'] + assert b1opts['ceph_rgw:access_key'] == 'AUTO_FETCHED_ACCESS_KEY' + assert b1opts['ceph_rgw:secret_access_key'] == 'AUTO_FETCHED_SECRET_KEY' + assert b2opts['ceph_rgw:access_key'] == 'AUTO_FETCHED_ACCESS_KEY' + assert b2opts['ceph_rgw:secret_access_key'] == 'AUTO_FETCHED_SECRET_KEY'