1
- import typing
1
+ from typing import List
2
2
from pathlib import Path
3
3
import warnings
4
4
from datetime import datetime
@@ -23,8 +23,7 @@ class ContainersAlreadyExist(Exception):
23
23
24
24
25
25
class NetworkInfo :
26
- def __init__ (self , container_port : typing .Text ,
27
- hostname : typing .Text , host_port : int ,):
26
+ def __init__ (self , container_port : str , hostname : str , host_port : int ,):
28
27
"""
29
28
Container for info about how to connect to a service exposed by a
30
29
Docker container.
@@ -39,7 +38,7 @@ def __init__(self, container_port: typing.Text,
39
38
self .host_port = host_port
40
39
41
40
42
- def create_network_info_for_container (container ):
41
+ def create_network_info_for_container (container : Container ):
43
42
"""
44
43
Generates :py:class:`NetworkInfo` instances corresponding to all available
45
44
port bindings in a container
@@ -140,7 +139,7 @@ def docker_project(self, request):
140
139
return project
141
140
142
141
@classmethod
143
- def generate_scoped_containers_fixture (cls , scope ):
142
+ def generate_scoped_containers_fixture (cls , scope : str ):
144
143
"""
145
144
Create scoped fixtures that retrieve or spin up all containers, and add
146
145
network info objects to containers and then yield the containers for
@@ -152,21 +151,20 @@ def generate_scoped_containers_fixture(cls, scope):
152
151
@pytest .fixture (scope = scope )
153
152
def scoped_containers_fixture (docker_project : Project , request ):
154
153
now = datetime .utcnow ()
155
- if request .config .getoption ("--use-running-containers" ):
156
- containers = docker_project .containers () # type: typing.List[Container]
157
- else :
154
+ if not request .config .getoption ("--use-running-containers" ):
158
155
if any (docker_project .containers ()):
159
156
raise ContainersAlreadyExist (
160
157
'pytest-docker-compose tried to start containers but there are'
161
158
' already running containers: %s, you probably scoped your'
162
159
' tests wrong' % docker_project .containers ())
163
- containers = docker_project .up () # type: typing. List[Container]
160
+ containers = docker_project .up () # type: List[Container]
164
161
if not containers :
165
162
raise ValueError ("`docker-compose` didn't launch any containers!" )
166
163
167
- yield ContainerDict (docker_project )
164
+ container_dict = ContainerDict (docker_project )
165
+ yield container_dict
168
166
169
- for container in sorted (containers , key = lambda c : c .name ):
167
+ for container in sorted (container_dict . values () , key = lambda c : c .name ):
170
168
header = "Logs from {name}:" .format (name = container .name )
171
169
print (header , '\n ' , "=" * len (header ))
172
170
print (container .logs (since = now ).decode ("utf-8" , errors = "replace" )
@@ -185,16 +183,17 @@ def scoped_containers_fixture(docker_project: Project, request):
185
183
186
184
187
185
class ContainerDict (dict ):
188
- """A read-only dictionary that re-fetches containers from a docker-compose
189
- project dynamically every time they are accessed."""
190
-
186
+ """
187
+ A read-only dictionary that re-fetches containers from a docker-compose
188
+ project dynamically every time they are accessed.
189
+ """
191
190
def __init__ (self , docker_project : Project ) -> None :
192
191
super ().__init__ ()
193
192
self .docker_project = docker_project
194
193
for container in docker_project .containers ():
195
194
super ().__setitem__ (container .name , container )
196
195
197
- def __getitem__ (self , key ) :
196
+ def __getitem__ (self , key : str ) -> Container :
198
197
container = [container for container in self .docker_project .containers ()
199
198
if container .name == key ][0 ]
200
199
container .network_info = create_network_info_for_container (container )
0 commit comments