|
13 | 13 | import singlestoredb as s2 |
14 | 14 | from singlestoredb.management.job import Status |
15 | 15 | from singlestoredb.management.job import TargetType |
| 16 | +from singlestoredb.management.region import Region |
| 17 | +from singlestoredb.management.utils import NamedList |
16 | 18 |
|
17 | 19 |
|
18 | 20 | TEST_DIR = pathlib.Path(os.path.dirname(__file__)) |
@@ -1372,3 +1374,107 @@ def test_file_object(self): |
1372 | 1374 |
|
1373 | 1375 | # Cleanup |
1374 | 1376 | space.remove('obj_test_2.ipynb') |
| 1377 | + |
| 1378 | + |
| 1379 | +@pytest.mark.management |
| 1380 | +class TestRegions(unittest.TestCase): |
| 1381 | + """Test cases for region management.""" |
| 1382 | + |
| 1383 | + manager = None |
| 1384 | + |
| 1385 | + @classmethod |
| 1386 | + def setUpClass(cls): |
| 1387 | + """Set up the test environment.""" |
| 1388 | + cls.manager = s2.manage_regions() |
| 1389 | + |
| 1390 | + @classmethod |
| 1391 | + def tearDownClass(cls): |
| 1392 | + """Clean up the test environment.""" |
| 1393 | + cls.manager = None |
| 1394 | + |
| 1395 | + def test_list_regions(self): |
| 1396 | + """Test listing all regions.""" |
| 1397 | + regions = self.manager.list_regions() |
| 1398 | + |
| 1399 | + # Verify we get a NamedList |
| 1400 | + assert isinstance(regions, NamedList) |
| 1401 | + |
| 1402 | + # Verify we have at least one region |
| 1403 | + assert len(regions) > 0 |
| 1404 | + |
| 1405 | + # Verify region properties |
| 1406 | + region = regions[0] |
| 1407 | + assert isinstance(region, Region) |
| 1408 | + assert hasattr(region, 'id') |
| 1409 | + assert hasattr(region, 'name') |
| 1410 | + assert hasattr(region, 'provider') |
| 1411 | + |
| 1412 | + # Verify provider values |
| 1413 | + providers = {x.provider for x in regions} |
| 1414 | + assert 'Azure' in providers or 'GCP' in providers or 'AWS' in providers |
| 1415 | + |
| 1416 | + # Verify region can be accessed by name or ID |
| 1417 | + region_by_name = regions[region.name] |
| 1418 | + region_by_id = regions[region.id] |
| 1419 | + assert region_by_name == region_by_id |
| 1420 | + assert region_by_name.id == region.id |
| 1421 | + assert region_by_name.name == region.name |
| 1422 | + assert region_by_name.provider == region.provider |
| 1423 | + |
| 1424 | + def test_list_shared_tier_regions(self): |
| 1425 | + """Test listing shared tier regions.""" |
| 1426 | + regions = self.manager.list_shared_tier_regions() |
| 1427 | + |
| 1428 | + # Verify we get a NamedList |
| 1429 | + assert isinstance(regions, NamedList) |
| 1430 | + |
| 1431 | + # Verify region properties if we have any shared tier regions |
| 1432 | + if regions: |
| 1433 | + region = regions[0] |
| 1434 | + assert isinstance(region, Region) |
| 1435 | + assert hasattr(region, 'id') |
| 1436 | + assert hasattr(region, 'name') |
| 1437 | + assert hasattr(region, 'provider') |
| 1438 | + |
| 1439 | + # Verify provider values |
| 1440 | + providers = {x.provider for x in regions} |
| 1441 | + assert any(p in providers for p in ['Azure', 'GCP', 'AWS']) |
| 1442 | + |
| 1443 | + # Verify region can be accessed by name or ID |
| 1444 | + region_by_name = regions[region.name] |
| 1445 | + region_by_id = regions[region.id] |
| 1446 | + assert region_by_name == region_by_id |
| 1447 | + assert region_by_name.id == region.id |
| 1448 | + assert region_by_name.name == region.name |
| 1449 | + assert region_by_name.provider == region.provider |
| 1450 | + |
| 1451 | + def test_str_repr(self): |
| 1452 | + """Test string representation of regions.""" |
| 1453 | + regions = self.manager.list_regions() |
| 1454 | + if not regions: |
| 1455 | + self.skipTest('No regions available for testing') |
| 1456 | + |
| 1457 | + region = regions[0] |
| 1458 | + |
| 1459 | + # Test __str__ |
| 1460 | + s = str(region) |
| 1461 | + assert region.id in s |
| 1462 | + assert region.name in s |
| 1463 | + assert region.provider in s |
| 1464 | + |
| 1465 | + # Test __repr__ |
| 1466 | + assert repr(region) == str(region) |
| 1467 | + |
| 1468 | + def test_no_manager(self): |
| 1469 | + """Test behavior when manager is not available.""" |
| 1470 | + regions = self.manager.list_regions() |
| 1471 | + if not regions: |
| 1472 | + self.skipTest('No regions available for testing') |
| 1473 | + |
| 1474 | + region = regions[0] |
| 1475 | + region._manager = None |
| 1476 | + |
| 1477 | + # Verify from_dict class method |
| 1478 | + with self.assertRaises(s2.ManagementError) as cm: |
| 1479 | + Region.get_shared_tier_regions(None) |
| 1480 | + assert 'No workspace manager' in str(cm.exception) |
0 commit comments