Coverage for tests / test_download.py: 98%

1""" 

2Tests for the download functions. 

3""" 

4 

5import unittest 

6 

7import aiohttp 

8import pandas as pd 

9 

10import acspsuedo.query as apq 

11from acspsuedo.source.low.exceptions import APIException 

12from acspsuedo.source.geog import GeoScopeException 

13from acspsuedo.datasets import ACS1, ACS5 

14from acspsuedo.fips.states import CA 

15 

16 

17 

18class TestDownload(unittest.TestCase): 

19 

20 def setUp(self) -> None: 

21 self.DATASET = ACS5 

22 self.YEAR = 2020 

23 self.VARIABLES = ['NAME', 'GEO_ID', 'B25058_001E'] 

24 

25 self.CONTENT = [ 

26 ['NAME', 'GEO_ID', 'STATE', 'YEAR', 'B25058_001E'], 

27 ['California', '06', '06', 2020, 1442] 

28 ] 

29 self.DF = pd.DataFrame( 

30 columns = self.CONTENT[0], 

31 data = self.CONTENT[1:] 

32 ) 

33 

34 self.ACS1_DATASET = ACS1 

35 

36 

37 def test_download_url(self): 

38 """ 

39 Formatted URL for sending queries. 

40 """ 

41 key_fmt = apq.api_key_config._get_api_key() 

42 URL = f'https://api.census.gov/data/2020/acs/acs5?get=B25058_001E,GEO_ID,NAME&for=state:06{key_fmt}' 

43 fmt_url, _, _ = apq._fmt_download_url( 

44 self.DATASET, 

45 self.YEAR, 

46 vars = self.VARIABLES, 

47 state = CA 

48 ) 

49 self.assertEqual(URL, fmt_url[0]) 

50 

51 

52 def test_download_return_type(self): 

53 """ 

54 Synchronous download. 

55 """ 

56 df = apq.download( 

57 self.DATASET, 

58 self.YEAR, 

59 variables = self.VARIABLES, 

60 state = CA 

61 ) 

62 

63 self.assertIsInstance(df, pd.DataFrame) 

64 self.assertEqual([self.CONTENT[1]], df.values.tolist()) 

65 

66 def test_download_geometries_non_existent_shpfile(self): 

67 """ 

68 A scenario where users query data alongside their geographic info, 

69 but TIGER shapefiles are not available. 

70 """ 

71 with self.assertWarns(UserWarning): 

72 apq.download( 

73 self.DATASET, 

74 2011, 

75 variables = self.VARIABLES, 

76 zip_code_tabulation_area = '*', 

77 include_geometries = True 

78 ) 

79 

80 def test_failed_download_unsupported_dataset(self): 

81 """ 

82 A failed download due to an unsupported dataset. 

83 """ 

84 with self.assertRaises(KeyError): 

85 df = apq.download( 

86 'foo/bar', 

87 self.YEAR, 

88 variables = self.VARIABLES, 

89 state = 'bar' 

90 ) 

91 

92 def test_failed_download_unsupported_year(self): 

93 """ 

94 A failed download due to an unsupported year. 

95 """ 

96 with self.assertRaises(APIException): 

97 df = apq.download( 

98 self.ACS1_DATASET, 

99 self.YEAR, 

100 variables = self.VARIABLES, 

101 state = CA 

102 ) 

103 

104 def test_failed_download_unsupported_geo_specifier(self): 

105 """ 

106 A failed download due to an unsupported geographic specifier. 

107 """ 

108 with self.assertRaises(GeoScopeException): 

109 df = apq.download( 

110 self.DATASET, 

111 self.YEAR, 

112 variables = self.VARIABLES, 

113 foo = 'bar' 

114 ) 

115 

116 def test_failed_download_incorrect_geo_specifier_value(self): 

117 """ 

118 A failed download due to an incorrect geographic specifier value. 

119 """ 

120 with self.assertRaises(APIException): 

121 df = apq.download( 

122 self.DATASET, 

123 self.YEAR, 

124 variables = self.VARIABLES, 

125 state = 'foo' 

126 ) 

127 

128 

129class TestAsyncDownload(unittest.IsolatedAsyncioTestCase): 

130 

131 def setUp(self) -> None: 

132 self.DATASET = ACS5 

133 self.YEAR = 2020 

134 self.VARIABLES = ['NAME', 'GEO_ID', 'B25058_001E'] 

135 

136 self.CONTENT = [ 

137 ['NAME', 'GEO_ID', 'STATE', 'YEAR', 'B25058_001E'], 

138 ['California', '06', '06', 2020, 1442] 

139 ] 

140 self.DF = pd.DataFrame( 

141 columns = self.CONTENT[0], 

142 data = self.CONTENT[1:] 

143 ) 

144 

145 async def asyncSetUp(self): 

146 self.ASYNC_SESSION = aiohttp.ClientSession() 

147 

148 async def asyncTearDown(self): 

149 await self.ASYNC_SESSION.close() 

150 

151 async def test_async_download_return_type(self): 

152 """ 

153 Asynchronous download. 

154 """ 

155 df = await apq.async_download( 

156 self.ASYNC_SESSION, 

157 self.DATASET, 

158 self.YEAR, 

159 variables = self.VARIABLES, 

160 state = CA 

161 ) 

162 

163 self.assertIsInstance(df, pd.DataFrame) 

164 self.assertEqual([self.CONTENT[1]], df.values.tolist()) 

165 

166 

167 

168 

169 

170if __name__ == '__main__': 

171 unittest.main()