You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

248 lines
9.0 KiB

  1. # Copyright 2015 Agile Business Group <http://www.agilebg.com>
  2. # Copyright 2015 Alessio Gerace <alesiso.gerace@agilebg.com>
  3. # Copyright 2016 Grupo ESOC Ingenieria de Servicios, S.L.U. - Jairo Llopis
  4. # Copyright 2016 LasLabs Inc.
  5. # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
  6. import os
  7. from contextlib import contextmanager
  8. from datetime import datetime, timedelta
  9. import mock
  10. from odoo import exceptions, tools
  11. from odoo.tests import common
  12. try:
  13. import pysftp
  14. except ImportError:
  15. pass
  16. model = 'odoo.addons.auto_backup.models.db_backup'
  17. class TestConnectionException(pysftp.ConnectionException):
  18. def __init__(self):
  19. super(TestConnectionException, self).__init__('test', 'test')
  20. class TestDbBackup(common.TransactionCase):
  21. def setUp(self):
  22. super(TestDbBackup, self).setUp()
  23. self.Model = self.env["db.backup"]
  24. @contextmanager
  25. def mock_assets(self):
  26. """ It provides mocked core assets """
  27. self.path_join_val = '/this/is/a/path'
  28. with mock.patch('%s.db' % model) as db:
  29. with mock.patch('%s.os' % model) as os:
  30. with mock.patch('%s.shutil' % model) as shutil:
  31. os.path.join.return_value = self.path_join_val
  32. yield {
  33. 'db': db,
  34. 'os': os,
  35. 'shutil': shutil,
  36. }
  37. @contextmanager
  38. def patch_filtered_sftp(self, record, mocks=None):
  39. """ It patches filtered record and provides a mock """
  40. if mocks is None:
  41. mocks = ['sftp_connection']
  42. mocks = {m: mock.DEFAULT for m in mocks}
  43. with mock.patch.object(record, 'filtered') as filtered:
  44. with mock.patch.object(record, 'backup_log'):
  45. with mock.patch.multiple(record, **mocks):
  46. filtered.side_effect = [], [record]
  47. yield filtered
  48. def new_record(self, method='sftp'):
  49. vals = {
  50. 'name': u'Têst backup',
  51. 'method': method,
  52. }
  53. if method == 'sftp':
  54. vals.update({
  55. 'sftp_host': 'test_host',
  56. 'sftp_port': '222',
  57. 'sftp_user': 'tuser',
  58. 'sftp_password': 'password',
  59. 'folder': '/folder/',
  60. })
  61. self.vals = vals
  62. return self.Model.create(vals)
  63. def test_compute_name_sftp(self):
  64. """ It should create proper SFTP URI """
  65. rec_id = self.new_record()
  66. self.assertEqual(
  67. 'sftp://%(user)s@%(host)s:%(port)s%(folder)s' % {
  68. 'user': self.vals['sftp_user'],
  69. 'host': self.vals['sftp_host'],
  70. 'port': self.vals['sftp_port'],
  71. 'folder': self.vals['folder'],
  72. },
  73. rec_id.name,
  74. )
  75. def test_check_folder(self):
  76. """ It should not allow recursive backups """
  77. rec_id = self.new_record('local')
  78. with self.assertRaises(exceptions.ValidationError):
  79. rec_id.write({
  80. 'folder': '%s/another/path' % tools.config.filestore(
  81. self.env.cr.dbname
  82. ),
  83. })
  84. @mock.patch('%s._' % model)
  85. def test_action_sftp_test_connection_success(self, _):
  86. """ It should raise connection succeeded warning """
  87. rec_id = self.new_record()
  88. with mock.patch.object(rec_id, 'sftp_connection'):
  89. with self.assertRaises(exceptions.Warning):
  90. rec_id.action_sftp_test_connection()
  91. _.assert_called_once_with("Connection Test Succeeded!")
  92. @mock.patch('%s._' % model)
  93. def test_action_sftp_test_connection_fail(self, _):
  94. """ It should raise connection fail warning """
  95. rec_id = self.new_record()
  96. with mock.patch.object(rec_id, 'sftp_connection') as conn:
  97. conn().__enter__.side_effect = TestConnectionException
  98. with self.assertRaises(exceptions.Warning):
  99. rec_id.action_sftp_test_connection()
  100. _.assert_called_once_with("Connection Test Failed!")
  101. def test_action_backup_local(self):
  102. """ It should backup local database """
  103. rec_id = self.new_record('local')
  104. filename = rec_id.filename(datetime.now())
  105. rec_id.action_backup()
  106. generated_backup = [f for f in os.listdir(rec_id.folder)
  107. if f >= filename]
  108. self.assertEqual(1, len(generated_backup))
  109. def test_action_backup_local_cleanup(self):
  110. """ Backup local database and cleanup old databases """
  111. rec_id = self.new_record('local')
  112. rec_id.days_to_keep = 1
  113. old_date = datetime.now() - timedelta(days=3)
  114. filename = rec_id.filename(old_date)
  115. rec_id.action_backup()
  116. generated_backup = [f for f in os.listdir(rec_id.folder)
  117. if f >= filename]
  118. self.assertEqual(2, len(generated_backup))
  119. filename = rec_id.filename(datetime.now())
  120. rec_id.action_backup()
  121. generated_backup = [f for f in os.listdir(rec_id.folder)
  122. if f >= filename]
  123. self.assertEqual(1, len(generated_backup))
  124. def test_action_backup_sftp_mkdirs(self):
  125. """ It should create remote dirs """
  126. rec_id = self.new_record()
  127. with self.mock_assets():
  128. with self.patch_filtered_sftp(rec_id):
  129. conn = rec_id.sftp_connection().__enter__()
  130. rec_id.action_backup()
  131. conn.makedirs.assert_called_once_with(rec_id.folder)
  132. def test_action_backup_sftp_mkdirs_conn_exception(self):
  133. """ It should guard from ConnectionException on remote.mkdirs """
  134. rec_id = self.new_record()
  135. with self.mock_assets():
  136. with self.patch_filtered_sftp(rec_id):
  137. conn = rec_id.sftp_connection().__enter__()
  138. conn.makedirs.side_effect = TestConnectionException
  139. rec_id.action_backup()
  140. # No error was raised, test pass
  141. self.assertTrue(True)
  142. def test_action_backup_sftp_remote_open(self):
  143. """ It should open remote file w/ proper args """
  144. rec_id = self.new_record()
  145. with self.mock_assets() as assets:
  146. with self.patch_filtered_sftp(rec_id):
  147. conn = rec_id.sftp_connection().__enter__()
  148. rec_id.action_backup()
  149. conn.open.assert_called_once_with(
  150. assets['os'].path.join(),
  151. 'wb'
  152. )
  153. def test_action_backup_all_search(self):
  154. """ It should search all records """
  155. rec_id = self.new_record()
  156. with mock.patch.object(rec_id, 'search'):
  157. rec_id.action_backup_all()
  158. rec_id.search.assert_called_once_with([])
  159. def test_action_backup_all_return(self):
  160. """ It should return result of backup operation """
  161. rec_id = self.new_record()
  162. with mock.patch.object(rec_id, 'search'):
  163. res = rec_id.action_backup_all()
  164. self.assertEqual(
  165. rec_id.search().action_backup(), res
  166. )
  167. @mock.patch('%s.pysftp' % model)
  168. def test_sftp_connection_init_passwd(self, pysftp):
  169. """ It should initiate SFTP connection w/ proper args and pass """
  170. rec_id = self.new_record()
  171. rec_id.sftp_connection()
  172. pysftp.Connection.assert_called_once_with(
  173. host=rec_id.sftp_host,
  174. username=rec_id.sftp_user,
  175. port=rec_id.sftp_port,
  176. password=rec_id.sftp_password,
  177. )
  178. @mock.patch('%s.pysftp' % model)
  179. def test_sftp_connection_init_key(self, pysftp):
  180. """ It should initiate SFTP connection w/ proper args and key """
  181. rec_id = self.new_record()
  182. rec_id.write({
  183. 'sftp_private_key': 'pkey',
  184. 'sftp_password': 'pkeypass',
  185. })
  186. rec_id.sftp_connection()
  187. pysftp.Connection.assert_called_once_with(
  188. host=rec_id.sftp_host,
  189. username=rec_id.sftp_user,
  190. port=rec_id.sftp_port,
  191. private_key=rec_id.sftp_private_key,
  192. private_key_pass=rec_id.sftp_password,
  193. )
  194. @mock.patch('%s.pysftp' % model)
  195. def test_sftp_connection_return(self, pysftp):
  196. """ It should return new sftp connection """
  197. rec_id = self.new_record()
  198. res = rec_id.sftp_connection()
  199. self.assertEqual(
  200. pysftp.Connection(), res,
  201. )
  202. def test_filename_default(self):
  203. """ It should not error and should return a .dump.zip file str """
  204. now = datetime.now()
  205. res = self.Model.filename(now)
  206. self.assertTrue(res.endswith(".dump.zip"))
  207. def test_filename_zip(self):
  208. """ It should return a dump.zip filename"""
  209. now = datetime.now()
  210. res = self.Model.filename(now, ext='zip')
  211. self.assertTrue(res.endswith(".dump.zip"))
  212. def test_filename_dump(self):
  213. """ It should return a dump filename"""
  214. now = datetime.now()
  215. res = self.Model.filename(now, ext='dump')
  216. self.assertTrue(res.endswith(".dump"))