FilesExpand file tree

 main

/

helpers.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

104 lines (78 loc) · 2.63 KB

 main

/

helpers.py

Top

File metadata and controls

• Code
• Blame

104 lines (78 loc) · 2.63 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

from __future__ import annotations

import datetime

import io

import json

import requests

import responses

from gitlab import base

MATCH_EMPTY_QUERY_PARAMS = [responses.matchers.query_param_matcher({})]

class FakeObject(base.RESTObject):

pass

class FakeObjectWithoutId(base.RESTObject):

_id_attr = None

class FakeObjectWithLongRepr(base.RESTObject):

_id_attr = None

_repr_attr = "test"

class OtherFakeObject(FakeObject):

_id_attr = "foo"

class FakeManager(base.RESTManager):

_path = "/tests"

_obj_cls = FakeObject

class FakeParent(FakeObject):

id = 42

class FakeManagerWithParent(base.RESTManager):

_path = "/tests/{test_id}/cases"

_obj_cls = FakeObject

_from_parent_attrs = {"test_id": "id"}

# NOTE: The function `httmock_response` and the class `Headers` is taken from

# https://github.com/patrys/httmock/ which is licensed under the Apache License, Version

# 2.0. Thus it is allowed to be used in this project.

# https://www.apache.org/licenses/GPL-compatibility.html

class Headers:

def __init__(self, res):

self.headers = res.headers

def get_all(self, name, failobj=None):

return self.getheaders(name)

def getheaders(self, name):

return [self.headers.get(name)]

def httmock_response(

status_code: int = 200,

content: str = "",

headers=None,

reason=None,

elapsed=0,

request: requests.models.PreparedRequest | None = None,

stream: bool = False,

http_vsn=11,

) -> requests.models.Response:

res = requests.Response()

res.status_code = status_code

if isinstance(content, (dict, list)):

content = json.dumps(content).encode("utf-8")

if isinstance(content, str):

content = content.encode("utf-8")

res._content = content

res._content_consumed = content

res.headers = requests.structures.CaseInsensitiveDict(headers or {})

res.encoding = requests.utils.get_encoding_from_headers(res.headers)

res.reason = reason

res.elapsed = datetime.timedelta(elapsed)

res.request = request

if hasattr(request, "url"):

res.url = request.url

if isinstance(request.url, bytes):

res.url = request.url.decode("utf-8")

if "set-cookie" in res.headers:

res.cookies.extract_cookies(

requests.cookies.MockResponse(Headers(res)),

requests.cookies.MockRequest(request),

)

if stream:

res.raw = io.BytesIO(content)

else:

res.raw = io.BytesIO(b"")

res.raw.version = http_vsn

# normally this closes the underlying connection,

# but we have nothing to free.

res.close = lambda *args, **kwargs: None

return res