1
+ import os
2
+ import sys
1
3
import tempfile
2
- from typing import TYPE_CHECKING , Optional
4
+ from typing import Optional
3
5
4
- from pytest import TempPathFactory
6
+ import pytest
5
7
from upath import UPath
6
8
7
9
from pytest_servers .exceptions import RemoteUnavailable
8
10
from pytest_servers .local import LocalPath
9
11
from pytest_servers .utils import random_string
10
12
11
- if TYPE_CHECKING :
12
- from pytest import FixtureRequest
13
-
14
13
15
14
class TempUPathFactory :
16
15
"""Factory for temporary directories with universal-pathlib and mocked servers""" # noqa: E501
17
16
18
17
mock_remotes = {
19
- "azure" : ("azurite" , "_azure_connection_string" ),
20
- "gcs" : ("fake_gcs_server" , "_gcs_endpoint_url" ),
21
- "s3" : ("s3_server" , "_s3_endpoint_url" ),
18
+ # remote: (fixture_name, config attribute name, requires docker)
19
+ "azure" : ("azurite" , "_azure_connection_string" , True ),
20
+ "gcs" : ("fake_gcs_server" , "_gcs_endpoint_url" , True ),
21
+ "s3" : ("s3_server" , "_s3_endpoint_url" , False ),
22
22
}
23
23
24
24
def __init__ (
@@ -27,38 +27,58 @@ def __init__(
27
27
azure_connection_string : Optional [str ] = None ,
28
28
gcs_endpoint_url : Optional [str ] = None ,
29
29
):
30
- self ._request : Optional ["FixtureRequest" ] = None
30
+ self ._request : Optional ["pytest. FixtureRequest" ] = None
31
31
32
- self ._local_path_factory : Optional ["TempPathFactory" ] = None
32
+ self ._local_path_factory : Optional ["pytest. TempPathFactory" ] = None
33
33
self ._azure_connection_string = azure_connection_string
34
34
self ._gcs_endpoint_url = gcs_endpoint_url
35
35
self ._s3_endpoint_url = s3_endpoint_url
36
36
37
37
@classmethod
38
38
def from_request (
39
- cls , request : "FixtureRequest" , * args , ** kwargs
39
+ cls , request : "pytest. FixtureRequest" , * args , ** kwargs
40
40
) -> "TempUPathFactory" :
41
41
"""Create a factory according to pytest configuration."""
42
42
tmp_upath_factory = cls (* args , ** kwargs )
43
- tmp_upath_factory ._local_path_factory = TempPathFactory . from_config (
44
- request .config , _ispytest = True
43
+ tmp_upath_factory ._local_path_factory = (
44
+ pytest . TempPathFactory . from_config ( request .config , _ispytest = True )
45
45
)
46
46
tmp_upath_factory ._request = request
47
47
48
48
return tmp_upath_factory
49
49
50
50
def _mock_remote_setup (self , fs : "str" ) -> None :
51
51
try :
52
- mock_remote_fixture , remote_config_name = self .mock_remotes [fs ]
52
+ (
53
+ mock_remote_fixture ,
54
+ remote_config_name ,
55
+ needs_docker ,
56
+ ) = self .mock_remotes [fs ]
53
57
except KeyError :
54
58
raise RemoteUnavailable (f"No mock remote available for fs: { fs } " )
55
59
56
60
if getattr (self , remote_config_name ): # remote is already configured
57
61
return
58
62
63
+ if needs_docker and os .environ .get ("CI" ):
64
+ if sys .platform == "win32" :
65
+ pytest .skip (
66
+ "disabled for Windows on Github Actions: "
67
+ "https://github.com/actions/runner-images/issues/1143"
68
+ )
69
+ elif sys .platform == "darwin" :
70
+ pytest .skip (
71
+ "disabled for MacOS on Github Actions: "
72
+ "https://github.com/actions/runner-images/issues/2150"
73
+ )
74
+
59
75
assert self ._request
60
- remote_config = self ._request .getfixturevalue (mock_remote_fixture )
61
- assert remote_config , f"Failed to setup remote for { fs } "
76
+ try :
77
+ remote_config = self ._request .getfixturevalue (mock_remote_fixture )
78
+ except pytest .FixtureLookupError :
79
+ raise RemoteUnavailable (
80
+ f'{ fs } : Failed to setup "{ mock_remote_fixture } " fixture'
81
+ )
62
82
setattr (self , remote_config_name , remote_config )
63
83
64
84
def mktemp (
0 commit comments