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'
--- /dev/null
+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'
projects / ceph.git / commitdiff