test_ffdnispdb.py 7.0 KB

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