123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
from collections import OrderedDict
|
|
from datetime import date
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from moto.core import BaseBackend, BackendDict, BaseModel
|
|
from .exceptions import (
|
|
ContainerNotFoundException,
|
|
ResourceNotFoundException,
|
|
PolicyNotFoundException,
|
|
)
|
|
|
|
|
|
class Container(BaseModel):
|
|
def __init__(self, **kwargs: Any):
|
|
self.arn = kwargs.get("arn")
|
|
self.name = kwargs.get("name")
|
|
self.endpoint = kwargs.get("endpoint")
|
|
self.status = kwargs.get("status")
|
|
self.creation_time = kwargs.get("creation_time")
|
|
self.lifecycle_policy: Optional[str] = None
|
|
self.policy: Optional[str] = None
|
|
self.metric_policy: Optional[str] = None
|
|
self.tags = kwargs.get("tags")
|
|
|
|
def to_dict(self, exclude: Optional[List[str]] = None) -> Dict[str, Any]:
|
|
data = {
|
|
"ARN": self.arn,
|
|
"Name": self.name,
|
|
"Endpoint": self.endpoint,
|
|
"Status": self.status,
|
|
"CreationTime": self.creation_time,
|
|
"Tags": self.tags,
|
|
}
|
|
if exclude:
|
|
for key in exclude:
|
|
del data[key]
|
|
return data
|
|
|
|
|
|
class MediaStoreBackend(BaseBackend):
|
|
def __init__(self, region_name: str, account_id: str):
|
|
super().__init__(region_name, account_id)
|
|
self._containers: Dict[str, Container] = OrderedDict()
|
|
|
|
def create_container(self, name: str, tags: Dict[str, str]) -> Container:
|
|
arn = f"arn:aws:mediastore:container:{name}"
|
|
container = Container(
|
|
arn=arn,
|
|
name=name,
|
|
endpoint=f"/{name}",
|
|
status="CREATING",
|
|
creation_time=date.today().strftime("%m/%d/%Y, %H:%M:%S"),
|
|
tags=tags,
|
|
)
|
|
self._containers[name] = container
|
|
return container
|
|
|
|
def delete_container(self, name: str) -> None:
|
|
if name not in self._containers:
|
|
raise ContainerNotFoundException()
|
|
del self._containers[name]
|
|
|
|
def describe_container(self, name: str) -> Container:
|
|
if name not in self._containers:
|
|
raise ResourceNotFoundException()
|
|
container = self._containers[name]
|
|
container.status = "ACTIVE"
|
|
return container
|
|
|
|
def list_containers(self) -> List[Dict[str, Any]]:
|
|
"""
|
|
Pagination is not yet implemented
|
|
"""
|
|
return [c.to_dict() for c in self._containers.values()]
|
|
|
|
def list_tags_for_resource(self, name: str) -> Optional[Dict[str, str]]:
|
|
if name not in self._containers:
|
|
raise ContainerNotFoundException()
|
|
tags = self._containers[name].tags
|
|
return tags
|
|
|
|
def put_lifecycle_policy(self, container_name: str, lifecycle_policy: str) -> None:
|
|
if container_name not in self._containers:
|
|
raise ResourceNotFoundException()
|
|
self._containers[container_name].lifecycle_policy = lifecycle_policy
|
|
|
|
def get_lifecycle_policy(self, container_name: str) -> str:
|
|
if container_name not in self._containers:
|
|
raise ResourceNotFoundException()
|
|
lifecycle_policy = self._containers[container_name].lifecycle_policy
|
|
if not lifecycle_policy:
|
|
raise PolicyNotFoundException()
|
|
return lifecycle_policy
|
|
|
|
def put_container_policy(self, container_name: str, policy: str) -> None:
|
|
if container_name not in self._containers:
|
|
raise ResourceNotFoundException()
|
|
self._containers[container_name].policy = policy
|
|
|
|
def get_container_policy(self, container_name: str) -> str:
|
|
if container_name not in self._containers:
|
|
raise ResourceNotFoundException()
|
|
policy = self._containers[container_name].policy
|
|
if not policy:
|
|
raise PolicyNotFoundException()
|
|
return policy
|
|
|
|
def put_metric_policy(self, container_name: str, metric_policy: str) -> None:
|
|
if container_name not in self._containers:
|
|
raise ResourceNotFoundException()
|
|
self._containers[container_name].metric_policy = metric_policy
|
|
|
|
def get_metric_policy(self, container_name: str) -> str:
|
|
if container_name not in self._containers:
|
|
raise ResourceNotFoundException()
|
|
metric_policy = self._containers[container_name].metric_policy
|
|
if not metric_policy:
|
|
raise PolicyNotFoundException()
|
|
return metric_policy
|
|
|
|
|
|
mediastore_backends = BackendDict(MediaStoreBackend, "mediastore")
|