test_ffdnispdb.py 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213
  1. from ffdnispdb import create_app, db, utils
  2. from ffdnispdb.models import ISP, CoveredArea
  3. from flask import Flask
  4. from flask.ext.sqlalchemy import SQLAlchemy
  5. import unittest
  6. import doctest
  7. import json
  8. def ISP_fixtures():
  9. isp1, isp2 = ISP(), ISP()
  10. isp1.name = 'Test ISP 1'
  11. isp1.shortname = 'ISP1'
  12. isp1.json_url = 'http://doesnt.exists/isp.json'
  13. isp1.json = {
  14. 'name': 'Test ISP 1',
  15. 'shortname': 'ISP1',
  16. 'coveredAreas': [
  17. {
  18. 'name': 'Some Area',
  19. 'technologies': ['dsl'],
  20. 'area': { "type": "Polygon", "coordinates": [[
  21. [ 0.889892578125, 48.32703913063476 ],
  22. [ 0.054931640625, 47.39834920035926 ],
  23. [ 0.142822265625, 46.837649560937464 ],
  24. [ 2.911376953125, 46.42271253466717 ],
  25. [ 4.39453125, 46.98774725646565 ],
  26. [ 4.4384765625, 48.52388120259336 ],
  27. [ 2.5927734375, 48.8936153614802 ],
  28. [ 0.889892578125, 48.32703913063476 ]
  29. ]]}
  30. },
  31. {
  32. 'name': 'Some Other Area',
  33. 'technologies': ['ftth'],
  34. },
  35. ],
  36. 'version': 0.1
  37. }
  38. isp2.name = 'Test ISP 2'
  39. isp2.shortname = 'ISP2'
  40. isp2.json_url = 'http://doesnt.exists/isp.json'
  41. isp2.json = {
  42. 'name': 'Test ISP 2',
  43. 'shortname': 'ISP2',
  44. 'coveredAreas': [
  45. {
  46. 'name': 'Middle of nowhere',
  47. 'technologies': ['dsl'],
  48. 'area': { "type": "Polygon", "coordinates": [[
  49. [ 0.889892578125, 48.32703913063476 ],
  50. [ 0.054931640625, 47.39834920035926 ],
  51. [ 0.142822265625, 46.837649560937464 ],
  52. [ 2.911376953125, 46.42271253466717 ],
  53. [ 4.39453125, 46.98774725646565 ],
  54. [ 4.4384765625, 48.52388120259336 ],
  55. [ 2.5927734375, 48.8936153614802 ],
  56. [ 0.889892578125, 48.32703913063476 ]
  57. ]]}
  58. },
  59. {
  60. 'name': 'Urban Area',
  61. 'technologies': ['ftth'],
  62. },
  63. ],
  64. 'version': 0.1
  65. }
  66. return (isp1, isp2)
  67. class TestCase(unittest.TestCase):
  68. def create_app(self, **kwargs):
  69. test_cfg={
  70. 'TESTING': True,
  71. 'WTF_CSRF_ENABLED': False,
  72. 'SQLALCHEMY_DATABASE_URI': 'sqlite://',
  73. }
  74. test_cfg.update(kwargs)
  75. return create_app(test_cfg)
  76. def setUp(self):
  77. self.app = self.create_app()
  78. self.client = self.app.test_client()
  79. with self.app.app_context():
  80. db.create_all()
  81. self._ctx = self.app.test_request_context()
  82. self._ctx.push()
  83. def tearDown(self):
  84. db.session.remove()
  85. db.drop_all()
  86. self._ctx.pop()
  87. def assertStatus(self, response, status_code):
  88. self.assertEqual(response.status_code, status_code)
  89. class TestForm(TestCase):
  90. def test_index(self):
  91. self.assertStatus(self.client.get('/'), 200)
  92. self.assertStatus(self.client.get('/isp/map_data.json'), 200)
  93. def test_projectform(self):
  94. resp = self.client.post('/isp/create/form', data={
  95. 'tech_email': 'admin@isp.com',
  96. 'name': 'Test',
  97. 'step': '1',
  98. 'covered_areas-0-name': 'Somewhere over the rainbow',
  99. 'covered_areas-0-technologies': 'dsl',
  100. 'covered_areas-0-technologies': 'ftth'
  101. })
  102. self.assertNotEqual(resp.location, None)
  103. self.assertEqual(ISP.query.filter_by(name='Test').count(), 1)
  104. self.assertEqual(
  105. CoveredArea.query.filter_by(name='Somewhere over the rainbow').count(), 1
  106. )
  107. class TestAPI(TestCase):
  108. def setUp(self):
  109. super(TestAPI, self).setUp()
  110. db.session.add_all(ISP_fixtures())
  111. db.session.commit()
  112. def check_isp_apiobj(self, isp, apiobj):
  113. self.assertEqual(apiobj['is_ffdn_member'], isp.is_ffdn_member)
  114. self.assertEqual(apiobj['json_url'], isp.json_url)
  115. self.assertEqual(apiobj['date_added'], utils.tosystemtz(isp.date_added).isoformat())
  116. self.assertEqual(apiobj['last_update'],
  117. utils.tosystemtz(isp.last_update_success).isoformat()
  118. if isp.last_update_success else None)
  119. self.assertEqual(apiobj['ispformat'], isp.json)
  120. def check_coveredarea_apiobj(self, ca, apiobj):
  121. self.assertEqual(apiobj['name'], ca.name)
  122. self.assertEqual(apiobj['geojson'], json.loads(ca.area_geojson)
  123. if ca.area_geojson is not None else None)
  124. def test_isps(self):
  125. c = self.client.get('/api/v1/isp/')
  126. self.assertStatus(c, 200)
  127. resp = json.loads(c.data)
  128. isps = ISP.query.filter_by(is_disabled=False)
  129. self.assertEqual(isps.count(),
  130. resp['total_items'])
  131. for isp in isps:
  132. m = filter(lambda i: i['id'] == isp.id, resp['isps'])[0]
  133. self.check_isp_apiobj(isp, m)
  134. def test_isp(self):
  135. isps = ISP.query.filter_by(is_disabled=False).all()
  136. victim = isps[0]
  137. c = self.client.get('/api/v1/isp/%d/'%victim.id)
  138. self.assertStatus(c, 200)
  139. resp = json.loads(c.data)
  140. self.check_isp_apiobj(victim, resp)
  141. victim.is_disabled = True
  142. db.session.commit()
  143. c = self.client.get('/api/v1/isp/%d/'%victim.id)
  144. self.assertStatus(c, 404)
  145. victim = isps[1]
  146. c = self.client.get('/api/v1/isp/%d/covered_areas/'%victim.id)
  147. resp = json.loads(c.data)
  148. for ca in victim.covered_areas:
  149. m = filter(lambda i: i['id'] == ca.id, resp['covered_areas'])[0]
  150. self.check_coveredarea_apiobj(ca, m)
  151. def test_urls(self):
  152. db_urls = ISP.query.filter(ISP.json_url != None).values('json_url')
  153. db_urls = [u[0] for u in db_urls]
  154. c = self.client.get('/api/v1/isp/export_urls/')
  155. self.assertStatus(c, 200)
  156. api_urls = map(lambda x: x['json_url'], json.loads(c.data)['isps'])
  157. self.assertEqual(len(api_urls), len(db_urls))
  158. for au in api_urls:
  159. self.assertIn(au, db_urls)
  160. def test_coveredarea(self):
  161. isp = ISP.query.filter_by(is_disabled=False).first()
  162. c = self.client.get('/api/v1/isp/%d/covered_areas/'%isp.id)
  163. self.assertStatus(c, 200)
  164. resp = json.loads(c.data)
  165. for ca in isp.covered_areas:
  166. m = filter(lambda i: i['id'] == ca.id, resp['covered_areas'])[0]
  167. self.check_coveredarea_apiobj(ca, m)
  168. def load_tests(loader, tests, ignore):
  169. from ffdnispdb import views, models, utils, forms, crawler, sessions
  170. tests.addTests(doctest.DocTestSuite(views))
  171. tests.addTests(doctest.DocTestSuite(models))
  172. tests.addTests(doctest.DocTestSuite(utils))
  173. tests.addTests(doctest.DocTestSuite(forms))
  174. tests.addTests(doctest.DocTestSuite(crawler))
  175. return tests
  176. if __name__ == '__main__':
  177. unittest.main()