Coverage for backend/tests/routers/test_data_tables.py: 100%

1""" 

2Test module for API router endpoints covering CRUD operations for JAM entities. 

3 

4This module contains comprehensive test classes for all API endpoints, organised into simple tables 

5(companies, keywords, aggregators, locations, files) and complex tables with relationships 

6(persons, jobs, job applications, interviews, job application updates). Each test class inherits 

7from CRUDTestBase to ensure consistent testing of standard CRUD operations, including authorisation, 

8validation, and error handling. Additional custom endpoint tests are included where applicable. 

9""" 

10 

11from app import schemas 

12from tests.conftest import CRUDTestBase 

13from tests.utils.table_data import ( 

14 COMPANY_DATA, 

15 LOCATION_DATA, 

16 PERSON_DATA, 

17 AGGREGATOR_DATA, 

18 KEYWORD_DATA, 

19 FILE_DATA, 

20 JOB_DATA, 

21 INTERVIEW_DATA, 

22 JOB_APPLICATION_UPDATE_DATA, 

23) 

24 

25 

26# ---------------------------------------------------- SIMPLE TABLES --------------------------------------------------- 

27 

28 

29class TestKeywordCRUD(CRUDTestBase): 

30 endpoint = "/keywords" 

31 schema = schemas.KeywordCreate 

32 out_schema = schemas.KeywordOut 

33 test_data = "test_keywords" 

34 create_data = KEYWORD_DATA 

35 update_data = { 

36 "id": 1, 

37 "name": "Updated Python", 

38 } 

39 

40 

41class TestAggregatorCRUD(CRUDTestBase): 

42 endpoint = "/aggregators" 

43 schema = schemas.AggregatorCreate 

44 out_schema = schemas.AggregatorOut 

45 test_data = "test_aggregators" 

46 create_data = AGGREGATOR_DATA 

47 update_data = { 

48 "name": "Updated LinkedIn", 

49 "url": "https://updated-linkedin.com", 

50 "id": 1, 

51 } 

52 

53 

54class TestCompanyCRUD(CRUDTestBase): 

55 endpoint = "/companies" 

56 schema = schemas.CompanyCreate 

57 out_schema = schemas.CompanyOut 

58 test_data = "test_companies" 

59 create_data = COMPANY_DATA 

60 update_data = { 

61 "name": "OXPV", 

62 "id": 1, 

63 } 

64 

65 def test_get_all_specific_company(self, authorised_clients, test_companies) -> None: 

66 response = authorised_clients[0].get(f"{self.endpoint}/?url=https://techcorp.com") 

67 assert response.status_code == 200 

68 

69 # The response should be a list, so check the first item 

70 companies = response.json() 

71 assert len(companies) > 0 

72 assert companies[0]["name"] == "Tech Corp" 

73 

74 def test_get_all_specific_id_not_owned(self, authorised_clients, test_companies) -> None: 

75 response = authorised_clients[1].get(f"{self.endpoint}/?id=1") 

76 assert response.status_code == 200 

77 assert len(response.json()) == 0 

78 

79 

80class TestLocationCRUD(CRUDTestBase): 

81 endpoint = "/locations" 

82 schema = schemas.LocationCreate 

83 out_schema = schemas.LocationOut 

84 test_data = "test_locations" 

85 create_data = LOCATION_DATA 

86 update_data = { 

87 "postcode": "OX5 1HN", 

88 "id": 1, 

89 } 

90 

91 

92class TestFileCRUD(CRUDTestBase): 

93 endpoint = "/files" 

94 schema = schemas.FileCreate 

95 out_schema = schemas.FileOut 

96 test_data = "test_files" 

97 create_data = FILE_DATA 

98 update_data = { 

99 "filename": "updated_john_doe_cv_2024.pdf", 

100 "size": 2560, 

101 "id": 1, 

102 } 

103 

104 def test_file_download_data_url_format(self, authorised_clients, test_files) -> None: 

105 """Test file download with Base64 data URL format""" 

106 

107 # Use existing test file instead of creating new one 

108 test_file = test_files[0] # Get first test file 

109 

110 # Download the file 

111 download_response = authorised_clients[0].get(f"{self.endpoint}/{test_file.id}/download") 

112 assert download_response.status_code == 200 

113 

114 # Verify content type and filename in headers 

115 assert download_response.headers["content-type"] in ["application/pdf", "text/plain; charset=utf-8"] 

116 assert f'filename="{test_file.filename}"' in download_response.headers["content-disposition"] 

117 

118 def test_file_download_plain_base64_format(self, authorised_clients, test_files) -> None: 

119 """Test file download with plain Base64 format (without data URL prefix)""" 

120 

121 # Use second test file if available, otherwise first 

122 test_file = test_files[1] if len(test_files) > 1 else test_files[0] 

123 

124 # Download the file 

125 download_response = authorised_clients[0].get(f"{self.endpoint}/{test_file.id}/download") 

126 assert download_response.status_code == 200 

127 

128 # Verify basic response structure 

129 assert len(download_response.content) > 0 

130 assert "content-disposition" in download_response.headers 

131 

132 def test_file_download_binary_content(self, authorised_clients, test_files) -> None: 

133 """Test file download with binary content (simulating image/PDF)""" 

134 

135 # Use third test file if available, otherwise first 

136 test_file = test_files[2] if len(test_files) > 2 else test_files[0] 

137 

138 # Download the file 

139 download_response = authorised_clients[0].get(f"{self.endpoint}/{test_file.id}/download") 

140 assert download_response.status_code == 200 

141 

142 # Verify content 

143 downloaded_content = download_response.content 

144 assert len(downloaded_content) > 0 

145 

146 # Verify headers 

147 assert "content-type" in download_response.headers 

148 assert f'filename="{test_file.filename}"' in download_response.headers["content-disposition"] 

149 

150 def test_file_download_not_found(self, authorised_clients) -> None: 

151 """Test file download with non-existent file ID""" 

152 

153 download_response = authorised_clients[0].get(f"{self.endpoint}/999/download") 

154 assert download_response.status_code == 404 

155 error_data = download_response.json() 

156 assert "File not found" in error_data["detail"] 

157 

158 def test_file_download_unauthorized(self, authorised_clients, test_files) -> None: 

159 """Test file download access control - users can only download their own files""" 

160 

161 # Use existing test file 

162 test_file = test_files[0] 

163 

164 # Try to download with second user (assuming test files belong to first user) 

165 download_response = authorised_clients[1].get(f"{self.endpoint}/{test_file.id}/download") 

166 assert download_response.status_code == 404 

167 error_data = download_response.json() 

168 assert "File not found" in error_data["detail"] 

169 

170 def test_file_download_empty_content(self, authorised_clients) -> None: 

171 """Test file download with empty/null content""" 

172 

173 # Create a file with empty content for this specific test case 

174 file_data = {"filename": "empty_file.txt", "content": "", "type": "text/plain", "size": 0} 

175 

176 # This might fail at creation if backend validates non-empty content 

177 # Adjust based on your actual validation rules 

178 create_response = authorised_clients[0].post(f"{self.endpoint}/", json=file_data) 

179 if create_response.status_code == 201: 

180 file_id = create_response.json()["id"] 

181 download_response = authorised_clients[0].get(f"{self.endpoint}/{file_id}/download") 

182 # Should either return empty content or handle gracefully 

183 assert download_response.status_code in [200, 404, 500] 

184 

185 

186# --------------------------------------------------- COMPLEX TABLES --------------------------------------------------- 

187 

188 

189class TestPersonCRUD(CRUDTestBase): 

190 endpoint = "/persons" 

191 schema = schemas.PersonCreate 

192 out_schema = schemas.PersonOut 

193 test_data = "test_persons" 

194 add_fixture = ["test_companies"] 

195 create_data = PERSON_DATA 

196 update_data = { 

197 "first_name": "OX", 

198 "id": 1, 

199 } 

200 

201 

202class TestJobCRUD(CRUDTestBase): 

203 endpoint = "/jobs" 

204 schema = schemas.JobCreate 

205 out_schema = schemas.JobOut 

206 test_data = "test_jobs" 

207 add_fixture = [ 

208 "test_persons", 

209 "test_locations", 

210 "test_keywords", 

211 "test_companies", 

212 "test_aggregators", 

213 "test_files", 

214 ] 

215 create_data = JOB_DATA 

216 update_data = { 

217 "title": "Updated title", 

218 "url": "https://updated-linkedin.com", 

219 "id": 1, 

220 } 

221 

222 

223class TestJobApplicationUpdateCRUD(CRUDTestBase): 

224 endpoint = "/jobapplicationupdates" 

225 schema = schemas.JobApplicationUpdateCreate 

226 out_schema = schemas.JobApplicationUpdateOut 

227 test_data = "test_job_application_updates" 

228 add_fixture = ["test_jobs"] 

229 create_data = JOB_APPLICATION_UPDATE_DATA 

230 update_data = { 

231 "id": 1, 

232 "note": "Updated note", 

233 } 

234 

235 

236class TestInterviewCRUD(CRUDTestBase): 

237 endpoint = "/interviews" 

238 schema = schemas.InterviewCreate 

239 out_schema = schemas.InterviewOut 

240 test_data = "test_interviews" 

241 add_fixture = ["test_jobs", "test_locations", "test_persons"] 

242 create_data = INTERVIEW_DATA 

243 update_data = { 

244 "job_id": 1, 

245 "note": "Interview went very well - positive feedback", 

246 "date": "2024-01-20T10:00:00", 

247 "id": 1, 

248 }