FilesExpand file tree

 tmp-generated

/

main_test.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

96 lines (71 loc) · 3.02 KB

 tmp-generated

/

main_test.py

Copy path

Top

File metadata and controls

• Code
• Blame

96 lines (71 loc) · 3.02 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

# Copyright 2018 Google LLC

#

# Licensed under the Apache License, Version 2.0 (the 'License');

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an 'AS IS' BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

from collections import UserDict

from datetime import datetime, timedelta, timezone

from unittest.mock import MagicMock, Mock, patch

import flask

import pytest

import requests

import responses

import main

# Create a fake "app" for generating test request contexts.

@pytest.fixture(scope="module")

def app():

return flask.Flask(__name__)

def test_lazy_globals(app):

with app.test_request_context():

main.lazy_globals(flask.request)

def test_scope_demo(app):

with app.test_request_context():

res = main.scope_demo(flask.request)

assert res == 'Per instance: 362880, per function: 45'

@responses.activate

def test_connection_pooling_200(app):

responses.add(responses.GET, 'http://example.com',

json={'status': 'OK'}, status=200)

with app.test_request_context():

main.connection_pooling(flask.request)

@responses.activate

def test_connection_pooling_404(app):

responses.add(responses.GET, 'http://example.com',

json={'error': 'not found'}, status=404)

with app.test_request_context():

with pytest.raises(requests.exceptions.HTTPError):

main.connection_pooling(flask.request)

def test_avoid_infinite_retries(capsys):

now = datetime.now(timezone.utc)

with patch('main.datetime', wraps=datetime) as datetime_mock:

datetime_mock.now = Mock(return_value=now)

old_context = UserDict()

old_context.timestamp = (now - timedelta(seconds=15)).isoformat()

old_context.event_id = 'old_event_id'

young_context = UserDict()

young_context.timestamp = (now - timedelta(seconds=5)).isoformat()

young_context.event_id = 'young_event_id'

main.avoid_infinite_retries(None, old_context)

out, _ = capsys.readouterr()

assert f"Dropped {old_context.event_id} (age 15000.0ms)" in out

main.avoid_infinite_retries(None, young_context)

out, _ = capsys.readouterr()

assert f"Processed {young_context.event_id} (age 5000.0ms)" in out

def test_retry_or_not():

with patch('main.error_client') as error_client_mock:

error_client_mock.report_exception = MagicMock()

event = Mock(data={})

main.retry_or_not(event, None)

assert error_client_mock.report_exception.call_count == 1

event.data = {'retry': True}

with pytest.raises(RuntimeError):

main.retry_or_not(event, None)

assert error_client_mock.report_exception.call_count == 2