Source Code for Module trac.upgrades.tests.db42

  1  # -*- coding: utf-8 -*- 
  2  # 
  3  # Copyright (C) 2016-2023 Edgewall Software 
  4  # All rights reserved. 
  5  # 
  6  # This software is licensed as described in the file COPYING, which 
  7  # you should have received as part of this distribution. The terms 
  8  # are also available at https://trac.edgewall.org/wiki/TracLicense. 
  9  # 
 10  # This software consists of voluntary contributions made by many 
 11  # individuals. For the exact contribution history, see the revision 
 12  # history and logs, available at https://trac.edgewall.org/. 
 13   
 14  import copy 
 15  import unittest 
 16   
 17  from trac.db.api import DatabaseManager 
 18  from trac.db.schema import Column, Index, Table 
 19  from trac.test import EnvironmentStub, mkdtemp 
 20  from trac.upgrades import db42 
 21  from trac.util.datefmt import datetime_now, to_utimestamp, utc 
 22   
 23  VERSION = 42 
 24   
 25  old_attachment_schema = \ 
 26      Table('attachment', key=('type', 'id', 'filename'))[ 
 27          Column('type'), 
 28          Column('id'), 
 29          Column('filename'), 
 30          Column('size', type='int'), 
 31          Column('time', type='int64'), 
 32          Column('description'), 
 33          Column('author'), 
 34          Column('ipnr')] 
 35   
 36  old_wiki_schema = \ 
 37      Table('wiki', key=('name', 'version'))[ 
 38          Column('name'), 
 39          Column('version', type='int'), 
 40          Column('time', type='int64'), 
 41          Column('author'), 
 42          Column('ipnr'), 
 43          Column('text'), 
 44          Column('comment'), 
 45          Column('readonly', type='int'), 
 46          Index(['time'])] 
 47  old_schema = (old_attachment_schema, old_wiki_schema) 
 48   
 49  new_attachment_schema = copy.deepcopy(old_attachment_schema) 
 50  new_attachment_schema.remove_columns(('ipnr',)) 
 51  new_wiki_schema = copy.deepcopy(old_wiki_schema) 
 52  new_wiki_schema.remove_columns(('ipnr',)) 
 53  new_schema = (new_attachment_schema, new_wiki_schema) 
 54   
 55   
 56 -class UpgradeTestCase(unittest.TestCase): 
 57   
 58 -    def setUp(self): 
 59          self.env = EnvironmentStub(path=mkdtemp()) 
 60          self.dbm = DatabaseManager(self.env) 
 61          with self.env.db_transaction: 
 62              self.dbm.drop_tables(new_schema) 
 63              self.dbm.create_tables(old_schema) 
 64              self.dbm.set_database_version(VERSION - 1) 
 65   
 66 -    def tearDown(self): 
 67          self.env.reset_db_and_disk() 
 68   
 69 -    def test_attachment_table_upgraded(self): 
 70          """The ipnr column is removed from the attachment table.""" 
 71          db42.do_upgrade(self.env, VERSION, None) 
 72   
 73          column_names = [col.name for col in new_attachment_schema.columns] 
 74          self.assertEqual(column_names, self.dbm.get_column_names('attachment')) 
 75   
 76 -    def test_wiki_table_upgraded(self): 
 77          """The ipnr column is removed from the wiki table.""" 
 78          db42.do_upgrade(self.env, VERSION, None) 
 79   
 80          column_names = [col.name for col in new_wiki_schema.columns] 
 81          self.assertEqual(column_names, self.dbm.get_column_names('wiki')) 
 82   
 83 -    def test_attachments_data_migrated(self): 
 84          """Attachment data is migrated on table upgrade.""" 
 85          now = to_utimestamp(datetime_now(utc)) 
 86          attachment_column_names = \ 
 87              [col.name for col in old_attachment_schema.columns] 
 88          attachment_data = ( 
 89              ('ticket', '1', 'file1', 10, now, 'desc1', 'user1', '::1'), 
 90              ('wiki', 'WikiStart', 'file2', 20, now, 'desc2', 'user2', '::2')) 
 91          self.dbm.insert_into_tables((('attachment', attachment_column_names, 
 92                                        attachment_data),)) 
 93   
 94          db42.do_upgrade(self.env, VERSION, None) 
 95   
 96          ipnr_col = attachment_column_names.index('ipnr') 
 97          i = 0 
 98          for i, data in enumerate(self.env.db_query(""" 
 99                  SELECT * FROM attachment ORDER BY type 
100                  """)): 
101              self.assertEqual(attachment_data[i][:ipnr_col] + 
102                               attachment_data[i][ipnr_col+1:], data) 
103          self.assertEqual(len(attachment_data), i+1) 
104   
105 -    def test_wiki_data_migrated(self): 
106          """Wiki data is migrated on table upgrade.""" 
107          now = to_utimestamp(datetime_now(utc)) 
108          wiki_column_names = \ 
109              [col.name for col in old_wiki_schema.columns] 
110          wiki_data = ( 
111              ('TracGuide', 2, now, 'user2', '::4', 'The guide', 'Edit', 0), 
112              ('WikiStart', 1, now, 'user1', '::3', 'The page', 'Init', 1)) 
113          self.dbm.insert_into_tables((('wiki', wiki_column_names, wiki_data),)) 
114   
115          db42.do_upgrade(self.env, VERSION, None) 
116   
117          ipnr_col = wiki_column_names.index('ipnr') 
118          i = 0 
119          for i, data in enumerate(self.env.db_query(""" 
120                  SELECT * FROM wiki ORDER BY name 
121                  """)): 
122              self.assertEqual(wiki_data[i][:ipnr_col] + 
123                               wiki_data[i][ipnr_col+1:], data) 
124          self.assertEqual(len(wiki_data), i+1) 
125   
126   
127 -def test_suite(): 
128      return unittest.makeSuite(UpgradeTestCase) 
129   
130  if __name__ == '__main__': 
131      unittest.main(defaultTest='test_suite') 
132