test_mongodb.py 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123
  1. # ============================================================================
  2. # This file is part of Pwman3.
  3. #
  4. # Pwman3 is free software; you can redistribute it and/or modify
  5. # it under the terms of the GNU General Public License, version 2
  6. # as published by the Free Software Foundation;
  7. #
  8. # Pwman3 is distributed in the hope that it will be useful,
  9. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  11. # GNU General Public License for more details.
  12. #
  13. # You should have received a copy of the GNU General Public License
  14. # along with Pwman3; if not, write to the Free Software
  15. # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
  16. # ============================================================================
  17. # Copyright (C) 2015 Oz Nahum Tiram <nahumoz@gmail.com>
  18. # ============================================================================
  19. import unittest
  20. import sys
  21. from .test_crypto_engine import give_key, DummyCallback
  22. if sys.version_info.major > 2: # pragma: no cover
  23. from urllib.parse import urlparse
  24. else: # pragma: no cover
  25. from urlparse import urlparse
  26. import pymongo
  27. from pwman.data.drivers.mongodb import MongoDB
  28. # use pwmantest
  29. # db.createUser(
  30. # {
  31. # user: "tester",
  32. # pwd: "12345678",
  33. # roles: [{ role: "dbAdmin", db: "pwmantest" },
  34. # { role: "readWrite", db: "pwmantest" },]
  35. # })
  36. from pwman.util.crypto_engine import CryptoEngine
  37. class TestMongoDB(unittest.TestCase):
  38. @classmethod
  39. def setUpClass(cls):
  40. u = u"mongodb://tester:12345678@localhost:27017/pwmantest"
  41. cls.db = MongoDB(u)
  42. cls.db._open()
  43. @classmethod
  44. def tearDownClass(cls):
  45. coll = cls.db._db['crypto']
  46. coll.drop()
  47. cls.db._db['counters'].drop()
  48. cls.db._db['nodes'].drop()
  49. cls.db.close()
  50. def test_1_con(self):
  51. self.assertIsInstance(self.db._con, pymongo.Connection)
  52. @unittest.skip("MongoDB creates collections on the fly")
  53. def test_2_create_collections(self):
  54. pass
  55. def test_3_load_key(self):
  56. self.db.savekey('SECRET$6$KEY')
  57. secretkey = self.db.loadkey()
  58. self.assertEqual(secretkey, u'SECRET$6$KEY')
  59. @unittest.skip("")
  60. def test_4_save_crypto(self):
  61. self.db.save_crypto_info("TOP", "SECRET")
  62. secretkey = self.db.loadkey()
  63. self.assertEqual(secretkey, 'TOP$6$SECRET')
  64. row = self.db.fetch_crypto_info()
  65. self.assertEqual(row, ('TOP', 'SECRET'))
  66. def test_5_add_node(self):
  67. innode = [u"TBONE", u"S3K43T", u"example.org", u"some note",
  68. [u"bartag", u"footag"]]
  69. self.db.add_node(innode)
  70. outnode = self.db.getnodes([1])[0]
  71. self.assertEqual(innode[:-1] + [t for t in innode[-1]], outnode[1:])
  72. def test_6_list_nodes(self):
  73. ret = self.db.listnodes()
  74. self.assertEqual(ret, [1])
  75. ret = self.db.listnodes("footag")
  76. self.assertEqual(ret, [1])
  77. def test_6a_list_tags(self):
  78. ret = self.db.listtags()
  79. self.assertListEqual(ret, ['bartag', 'footag'])
  80. def test_6b_get_nodes(self):
  81. ret = self.db.getnodes([1])
  82. retb = self.db.getnodes([])
  83. self.assertListEqual(ret, retb)
  84. @unittest.skip("tags are created in situ in mongodb")
  85. def test_7_get_or_create_tag(self):
  86. pass
  87. @unittest.skip("tags are removed with their node")
  88. def test_7a_clean_orphans(self):
  89. pass
  90. def test_8_remove_node(self):
  91. self.db.removenodes([1])
  92. n = self.db.listnodes()
  93. self.assertEqual(len(n), 0)
  94. @unittest.skip("No schema migration with mongodb")
  95. def test_9_check_db_version(self):
  96. pass
  97. if __name__ == '__main__':
  98. ce = CryptoEngine.get()
  99. ce.callback = DummyCallback()
  100. ce.changepassword(reader=give_key)
  101. unittest.main(verbosity=2, failfast=True)