Coverage for /home/ubuntu/f8s/python/f8s/test_app.py: 94%

1import json 

2 

3import flask 

4import flasgger as swg 

5 

6from f8s.extension import F8s 

7import f8s.tools as f8t 

8 

9 

10# API--------------------------------------------------------------------------- 

11API = flask.Blueprint('test', __name__, url_prefix='') 

12 

13 

14@API.route('/api/v1/get', methods=['GET']) 

15@swg.swag_from(dict( 

16 parameters=[], 

17 responses={ 

18 200: dict( 

19 description='test endpoint.', 

20 content='application/json', 

21 ), 

22 500: dict( 

23 description='Internal server error.', 

24 ) 

25 } 

26)) 

27def get(): 

28 # type: () -> flask.Response 

29 ''' 

30 Get test response. 

31 

32 Returns: 

33 Response: Flask Response instance. 

34 ''' 

35 return flask.Response( 

36 response=json.dumps(dict(message='Success')), 

37 mimetype='application/json' 

38 ) 

39 

40 

41@API.route('/api/v1/post', methods=['POST']) 

42@swg.swag_from(dict( 

43 parameters=[ 

44 dict( 

45 name='params', 

46 type='dict', 

47 description='Test params.', 

48 required=True, 

49 ) 

50 ], 

51 responses={ 

52 200: dict(description='JSON data', content='application/json'), 

53 500: dict(description='Internal server error.') 

54 } 

55)) 

56def post(): 

57 # type: () -> flask.Response 

58 ''' 

59 Post test data. 

60 

61 Returns: 

62 Response: Flask Response instance. 

63 ''' 

64 data = json.loads(flask.request.json) # type: ignore 

65 return flask.Response( 

66 response=json.dumps(data), 

67 mimetype='application/json' 

68 ) 

69 

70 

71@API.route('/api/v1/config', methods=['GET']) 

72@swg.swag_from(dict( 

73 parameters=[], 

74 responses={ 

75 200: dict(description='Config data', content='application/json'), 

76 500: dict(description='Internal server error.') 

77 } 

78)) 

79def config(): 

80 # type: () -> flask.Response 

81 ''' 

82 Get config data. 

83 

84 Returns: 

85 Response: Flask Response instance. 

86 ''' 

87 data = flask.current_app.config['demo'] 

88 return flask.Response( 

89 response=json.dumps(data), 

90 mimetype='application/json' 

91 ) 

92 

93 

94# EXTENSION--------------------------------------------------------------------- 

95class Demo(F8s): 

96 api = API 

97 

98 def validate(self, config): 

99 if 'foo' in config.keys(): 

100 assert config['foo'] == 'bar' 

101 

102 

103def live_probe(): 

104 # type: () -> dict 

105 ''' 

106 Liveness probe for kubernetes. 

107 ''' 

108 return dict(status='live') 

109 

110 

111def ready_probe(): 

112 # type: () -> dict 

113 ''' 

114 Readiness probe for kubernetes. 

115 ''' 

116 return dict(status='ready') 

117 

118 

119# APP--------------------------------------------------------------------------- 

120def get_app(): 

121 return f8t.get_app([Demo()], live_probe, ready_probe) 

122 

123 

124app = get_app()