|
| 1 | +from unittest.mock import mock_open, patch |
| 2 | + |
| 3 | +from ctorm.config import ConfigError, CtormBucket, CtormConfig |
| 4 | + |
| 5 | + |
| 6 | +def test_ctorm_bucket_initialization(): |
| 7 | + bucket = CtormBucket(bucketname="test_bucket", share=50, ummg_prefix="test_prefix/") |
| 8 | + |
| 9 | + assert bucket.bucketname == "test_bucket" |
| 10 | + assert bucket.share == 50 |
| 11 | + assert bucket.ummg_prefix == "test_prefix/" |
| 12 | + assert bucket.next_cont_token is None |
| 13 | + |
| 14 | + |
| 15 | +def test_ctorm_config_from_file_success(): |
| 16 | + mock_data = """ |
| 17 | + [ctorm] |
| 18 | + source_buckets = [{"bucketname": "test_bucket", "share": 50}] |
| 19 | + granules_sqs_queue_url = "https://sqs.queue.url/" |
| 20 | + """ |
| 21 | + with patch("builtins.open", mock_open(read_data=mock_data)), patch("tomllib.load") as mock_toml: |
| 22 | + mock_toml.return_value = { |
| 23 | + "ctorm": { |
| 24 | + "source_buckets": [{"bucketname": "test_bucket", "share": 50}], |
| 25 | + "granules_sqs_queue_url": "https://sqs.queue.url/", |
| 26 | + } |
| 27 | + } |
| 28 | + config = CtormConfig.from_file("test.cfg") |
| 29 | + |
| 30 | + assert config.granules_sqs_queue_url == "https://sqs.queue.url/" |
| 31 | + assert len(config.source_buckets) == 1 |
| 32 | + assert isinstance(config.source_buckets[0], CtormBucket) |
| 33 | + assert config.source_buckets[0].bucketname == "test_bucket" |
| 34 | + |
| 35 | + |
| 36 | +def test_ctorm_config_from_file_missing_section(): |
| 37 | + mock_data = "{}" |
| 38 | + with patch("builtins.open", mock_open(read_data=mock_data)), patch("tomllib.load") as mock_toml: |
| 39 | + mock_toml.return_value = {} |
| 40 | + try: |
| 41 | + CtormConfig.from_file("test.cfg") |
| 42 | + except ConfigError as e: |
| 43 | + assert str(e) == "No 'ctorm' section in config file" |
0 commit comments