Coverage for tests/docker_mock.py: 99%

1import random 

2from collections import defaultdict 

3 

4import pendulum 

5import docker 

6 

7 

8def cid(short=False): 

9 n_chars = 12 if short else 64 

10 return "".join(random.sample("0123456789abcdef" * 10, n_chars)) 

11 

12 

13class Network: 

14 def __init__(self, **kwargs): 

15 self.__dict__.update(kwargs) 

16 

17 def remove(self, **_): 

18 pass 

19 

20 

21class Networks: 

22 nets = {} 

23 

24 def list(self, names): 

25 return list(filter(None, [self.nets.get(name) for name in names])) 

26 

27 def create(self, **kwargs): 

28 name = kwargs.get("name") 

29 self.nets[name] = Network(**kwargs) 

30 return name 

31 

32 def get(self, name): 

33 return self.nets[name] 

34 

35 

36class Container: 

37 def __init__(self, **kwargs): 

38 self.id = cid() 

39 self.__dict__.update(kwargs) 

40 self.FinishedAt = "0001-01-01T00:00:00Z" 

41 self.ExitCode = 0 

42 self.attrs = { 

43 "State": { 

44 "StartedAt": getattr(self, "StartedAt", None), 

45 "FinishedAt": getattr(self, "FinishedAt", None), 

46 } 

47 } 

48 

49 def logs(self): 

50 return b"" 

51 

52 def stop(self, **_): 

53 self.FinishedAt = str(pendulum.now()) 

54 self.attrs["State"]["FinishedAt"] = self.FinishedAt 

55 self.ExitCode = 0 

56 

57 def remove(self, **_): 

58 Containers.boxes.pop(self.id, None) 

59 

60 

61class Containers: 

62 boxes = {} 

63 

64 def get(self, container_id): 

65 return self.boxes[container_id] 

66 

67 def run(self, **kwargs): 

68 kwargs["StartedAt"] = str(pendulum.now()) 

69 c = Container(**kwargs) 

70 self.boxes[c.id] = c 

71 return c 

72 

73 def list(self, **_): 

74 return list(Containers.boxes.values()) 

75 

76 

77class Docker: 

78 networks = Networks() 

79 containers = Containers() 

80 

81 

82class APIClient: 

83 max_running = {} 

84 reported_running = defaultdict(int) 

85 

86 def inspect_container(self, container_id): 

87 if container_id not in self.max_running: 

88 self.max_running[container_id] = random.choice(range(3, 11)) 

89 self.reported_running[container_id] += 1 

90 is_running = ( 

91 self.reported_running[container_id] <= self.max_running[container_id] 

92 ) 

93 container = Containers.boxes[container_id] 

94 return { 

95 "State": { 

96 "Running": is_running, 

97 "ExitCode": container.ExitCode, 

98 "StartedAt": container.StartedAt, 

99 "FinishedAt": container.FinishedAt, 

100 } 

101 } 

102 

103 

104def from_env(): 

105 return Docker() 

106 

107 

108docker.from_env = from_env 

109docker.APIClient = APIClient