aetherscale

[unmaintained] code for a cloud provider tutorial
Log | Files | Refs | README | LICENSE

conftest.py (2407B)

      1 from pathlib import Path
      2 import pytest
      3 from typing import Optional, List
      4 
      5 from aetherscale.services import ServiceManager
      6 import aetherscale.timing
      7 
      8 
      9 @pytest.fixture
     10 def tmppath(tmpdir):
     11     yield Path(tmpdir)
     12 
     13 
     14 @pytest.fixture
     15 def timeout():
     16     return aetherscale.timing.timeout
     17 
     18 
     19 @pytest.fixture
     20 def mock_service_manager():
     21     class MockServiceManager(ServiceManager):
     22         def __init__(self):
     23             self.services = set()
     24             self.started_services = set()
     25             self.enabled_services = set()
     26 
     27         def install_service(self, config_file: Path, service_name: str) -> bool:
     28             self.services.add(service_name)
     29             return True
     30 
     31         def install_simple_service(
     32                 self, command: str, service_name: str,
     33                 description: Optional[str] = None) -> bool:
     34             self.services.add(service_name)
     35             return True
     36 
     37         def uninstall_service(self, service_name: str) -> bool:
     38             try:
     39                 self.services.remove(service_name)
     40             except KeyError:
     41                 # should not fail if was already uninstalled
     42                 pass
     43 
     44             return True
     45 
     46         def start_service(self, service_name: str) -> bool:
     47             self.started_services.add(service_name)
     48             return True
     49 
     50         def stop_service(self, service_name: str) -> bool:
     51             try:
     52                 self.started_services.remove(service_name)
     53             except KeyError:
     54                 # should not fail if was already stopped
     55                 pass
     56 
     57             return True
     58 
     59         def restart_service(self, service_name: str) -> bool:
     60             return True
     61 
     62         def enable_service(self, service_name: str) -> bool:
     63             self.enabled_services.add(service_name)
     64             return True
     65 
     66         def disable_service(self, service_name: str) -> bool:
     67             try:
     68                 self.enabled_services.remove(service_name)
     69             except KeyError:
     70                 # should not fail if was already disabled
     71                 pass
     72 
     73             return True
     74 
     75         def service_is_running(self, service_name: str) -> bool:
     76             return service_name in self.started_services
     77 
     78         def service_exists(self, service_name: str) -> bool:
     79             return service_name in self.services
     80 
     81         def list_services(self) -> List[str]:
     82             return self.services
     83 
     84     return MockServiceManager()