test_mongodb.py 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117
  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. from .test_crypto_engine import give_key, DummyCallback
  21. import pymongo
  22. from pwman.data.drivers.mongodb import MongoDB
  23. # use pwmantest
  24. # db.createUser(
  25. # {
  26. # user: "tester",
  27. # pwd: "12345678",
  28. # roles: [{ role: "dbAdmin", db: "pwmantest" },
  29. # { role: "readWrite", db: "pwmantest" },]
  30. # })
  31. from pwman.util.crypto_engine import CryptoEngine
  32. class TestMongoDB(unittest.TestCase):
  33. @classmethod
  34. def setUpClass(cls):
  35. u = u"mongodb://tester:12345678@localhost:27017/pwmantest"
  36. cls.db = MongoDB(u)
  37. cls.db._open()
  38. @classmethod
  39. def tearDownClass(cls):
  40. coll = cls.db._db['crypto']
  41. coll.drop()
  42. cls.db._db['counters'].drop()
  43. cls.db._db['nodes'].drop()
  44. cls.db.close()
  45. def test_1_con(self):
  46. self.assertIsInstance(self.db._con, pymongo.Connection)
  47. @unittest.skip("MongoDB creates collections on the fly")
  48. def test_2_create_collections(self):
  49. pass
  50. def test_3_load_key(self):
  51. self.db.savekey('SECRET$6$KEY')
  52. secretkey = self.db.loadkey()
  53. self.assertEqual(secretkey, u'SECRET$6$KEY')
  54. @unittest.skip("")
  55. def test_4_save_crypto(self):
  56. self.db.save_crypto_info("TOP", "SECRET")
  57. secretkey = self.db.loadkey()
  58. self.assertEqual(secretkey, 'TOP$6$SECRET')
  59. row = self.db.fetch_crypto_info()
  60. self.assertEqual(row, ('TOP', 'SECRET'))
  61. def test_5_add_node(self):
  62. innode = [u"TBONE", u"S3K43T", u"example.org", u"some note",
  63. [u"bartag", u"footag"]]
  64. self.db.add_node(innode)
  65. outnode = self.db.getnodes([1])[0]
  66. self.assertEqual(innode[:-1] + [t for t in innode[-1]], outnode[1:])
  67. def test_6_list_nodes(self):
  68. ret = self.db.listnodes()
  69. self.assertEqual(ret, [1])
  70. ret = self.db.listnodes("footag")
  71. self.assertEqual(ret, [1])
  72. def test_6a_list_tags(self):
  73. ret = self.db.listtags()
  74. self.assertListEqual(ret, ['bartag', 'footag'])
  75. def test_6b_get_nodes(self):
  76. ret = self.db.getnodes([1])
  77. retb = self.db.getnodes([])
  78. self.assertListEqual(ret, retb)
  79. @unittest.skip("tags are created in situ in mongodb")
  80. def test_7_get_or_create_tag(self):
  81. pass
  82. @unittest.skip("tags are removed with their node")
  83. def test_7a_clean_orphans(self):
  84. pass
  85. def test_8_remove_node(self):
  86. self.db.removenodes([1])
  87. n = self.db.listnodes()
  88. self.assertEqual(len(n), 0)
  89. @unittest.skip("No schema migration with mongodb")
  90. def test_9_check_db_version(self):
  91. pass
  92. if __name__ == '__main__':
  93. ce = CryptoEngine.get()
  94. ce.callback = DummyCallback()
  95. ce.changepassword(reader=give_key)
  96. unittest.main(verbosity=2, failfast=True)