| # -*- coding: utf-8 -*- |
| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| from sqlite3 import OperationalError |
| from contextlib import contextmanager |
| import os |
| import shutil |
| import sys |
| import tempfile |
| import uuid |
| if sys.version_info < (2, 7): |
| import unittest2 as unittest |
| else: |
| import unittest |
| |
| from trac.attachment import Attachment, AttachmentAdmin |
| from trac.core import Component, implements |
| from trac.db import DatabaseManager |
| from trac.db.schema import Table, Column |
| from trac.env import IEnvironmentSetupParticipant |
| from trac.test import Environment |
| from trac.ticket import Ticket |
| from trac.wiki import WikiPage |
| |
| from multiproduct.api import MultiProductSystem |
| from multiproduct.env import ProductEnvironment |
| from multiproduct.model import Product |
| |
| BLOODHOUND_TABLES = ( |
| 'bloodhound_product', |
| 'bloodhound_productconfig', |
| 'bloodhound_productresourcemap', |
| ) |
| |
| TABLES_WITH_PRODUCT_FIELD = ( |
| 'ticket', 'ticket_change', 'ticket_custom', 'attachment', 'component', |
| 'milestone', 'wiki', 'report', |
| 'version', 'enum', 'permission', 'system', |
| ) |
| |
| |
| class EnvironmentUpgradeTestCase(unittest.TestCase): |
| def setUp(self, options=()): |
| self.env_path = tempfile.mkdtemp('multiproduct-tempenv') |
| self.env = Environment(self.env_path, create=True, options=options) |
| DummyPlugin.version = 1 |
| |
| def tearDown(self): |
| shutil.rmtree(self.env_path) |
| |
| def test_can_upgrade_environment_with_multi_product_disabled(self): |
| self.env.upgrade() |
| |
| # Multiproduct was not enabled so multiproduct tables should not exist |
| for table in BLOODHOUND_TABLES: |
| with self.assertFailsWithMissingTable(): |
| self.env.db_direct_query("SELECT * FROM %s" % table) |
| |
| for table in TABLES_WITH_PRODUCT_FIELD: |
| with self.assertFailsWithMissingColumn(): |
| self.env.db_direct_query("SELECT product FROM %s" % table) |
| |
| def test_upgrade_creates_multi_product_tables_and_adds_product_column(self): |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| with self.env.db_direct_transaction as db: |
| for table in BLOODHOUND_TABLES: |
| db("SELECT * FROM %s" % table) |
| |
| for table in TABLES_WITH_PRODUCT_FIELD: |
| db("SELECT product FROM %s" % table) |
| |
| def test_upgrade_creates_default_product(self): |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| products = Product.select(self.env) |
| self.assertEqual(len(products), 1) |
| |
| def test_upgrade_moves_tickets_and_related_objects_to_default_prod(self): |
| self._add_custom_field('custom_field') |
| with self.env.db_direct_transaction as db: |
| db("""INSERT INTO ticket (id) VALUES (1)""") |
| db("""INSERT INTO attachment (type, id, filename) |
| VALUES ('ticket', '1', '')""") |
| db("""INSERT INTO ticket_custom (ticket, name, value) |
| VALUES (1, 'custom_field', '42')""") |
| db("""INSERT INTO ticket_change (ticket, time, field) |
| VALUES (1, 42, 'summary')""") |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| with self.product('@'): |
| ticket = Ticket(self.env, 1) |
| attachments = list(Attachment.select(self.env, |
| ticket.resource.realm, |
| ticket.resource.id)) |
| self.assertEqual(len(attachments), 1) |
| self.assertEqual(ticket['custom_field'], '42') |
| changes = ticket.get_changelog() |
| self.assertEqual(len(changes), 3) |
| |
| def test_upgrade_moves_custom_wikis_to_default_product(self): |
| with self.env.db_direct_transaction as db: |
| db("""INSERT INTO wiki (name, version) VALUES ('MyPage', 1)""") |
| db("""INSERT INTO attachment (type, id, filename) |
| VALUES ('wiki', 'MyPage', '')""") |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| with self.env.db_direct_transaction as db: |
| self.assertEqual( |
| len(db("""SELECT * FROM wiki WHERE product='@'""")), 1) |
| self.assertEqual( |
| len(db("""SELECT * FROM attachment |
| WHERE product='@' |
| AND type='wiki'""")), 1) |
| |
| def test_upgrade_moves_system_wikis_to_products(self): |
| with self.env.db_direct_transaction as db: |
| db("""INSERT INTO wiki (name, version) VALUES ('WikiStart', 1)""") |
| db("""INSERT INTO attachment (type, id, filename) |
| VALUES ('wiki', 'WikiStart', '')""") |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| with self.env.db_direct_transaction as db: |
| self.assertEqual( |
| len(db("""SELECT * FROM wiki WHERE product='@'""")), 1) |
| self.assertEqual( |
| len(db("""SELECT * FROM attachment |
| WHERE product='@' |
| AND type='wiki'""")), 1) |
| self.assertEqual( |
| len(db("""SELECT * FROM wiki WHERE product=''""")), 0) |
| self.assertEqual( |
| len(db("""SELECT * FROM attachment |
| WHERE product='' |
| AND type='wiki'""")), 0) |
| |
| def test_upgrade_copies_content_of_system_tables_to_all_products(self): |
| mp = MultiProductSystem(self.env) |
| with self.env.db_direct_transaction as db: |
| mp._add_column_product_to_ticket(db) |
| mp._create_multiproduct_tables(db) |
| mp._update_db_version(db, 1) |
| for i in range(1, 6): |
| db("""INSERT INTO bloodhound_product (prefix, name) |
| VALUES ('p%d', 'Product 1')""" % i) |
| for table in ('component', 'milestone', 'enum', 'version', |
| 'permission', 'report'): |
| db("""DELETE FROM %s""" % table) |
| db("""INSERT INTO component (name) VALUES ('foobar')""") |
| db("""INSERT INTO milestone (name) VALUES ('foobar')""") |
| db("""INSERT INTO version (name) VALUES ('foobar')""") |
| db("""INSERT INTO enum (type, name) VALUES ('a', 'b')""") |
| db("""INSERT INTO permission VALUES ('x', 'TICKET_VIEW')""") |
| db("""INSERT INTO report (title) VALUES ('x')""") |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| with self.env.db_direct_transaction as db: |
| for table in ('component', 'milestone', 'version', 'enum', |
| 'report'): |
| rows = db("SELECT * FROM %s" % table) |
| self.assertEqual( |
| len(rows), 6, |
| "Wrong number of lines in %s (%d instead of %d)\n%s" |
| % (table, len(rows), 6, rows)) |
| for table in ('permission',): |
| # Permissions also hold rows for global product. |
| rows = db("SELECT * FROM %s WHERE username='x'" % table) |
| self.assertEqual( |
| len(rows), 7, |
| "Wrong number of lines in %s (%d instead of %d)\n%s" |
| % (table, len(rows), 7, rows)) |
| |
| def test_upgrading_database_moves_attachment_to_correct_product(self): |
| ticket = self.insert_ticket('ticket') |
| wiki = self.insert_wiki('MyWiki') |
| attachment = self._create_file_with_content('Hello World!') |
| self.add_attachment(ticket.resource, attachment) |
| self.add_attachment(wiki.resource, attachment) |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| with self.product('@'): |
| attachments = list( |
| Attachment.select(self.env, 'ticket', ticket.id)) |
| attachments.extend( |
| Attachment.select(self.env, 'wiki', wiki.name)) |
| self.assertEqual(len(attachments), 2) |
| for attachment in attachments: |
| self.assertEqual(attachment.open().read(), 'Hello World!') |
| |
| def test_can_upgrade_database_with_ticket_attachment_with_text_ids(self): |
| with self.env.db_direct_transaction as db: |
| db("""INSERT INTO attachment (id, type, filename) |
| VALUES ('abc', 'ticket', '')""") |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| def test_can_upgrade_database_with_orphaned_attachments(self): |
| with self.env.db_direct_transaction as db: |
| db("""INSERT INTO attachment (id, type, filename) |
| VALUES ('5', 'ticket', '')""") |
| db("""INSERT INTO attachment (id, type, filename) |
| VALUES ('MyWiki', 'wiki', '')""") |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| def test_can_upgrade_multi_product_from_v1(self): |
| mp = MultiProductSystem(self.env) |
| with self.env.db_direct_transaction as db: |
| mp._add_column_product_to_ticket(db) |
| mp._create_multiproduct_tables(db) |
| mp._update_db_version(db, 1) |
| |
| db("""INSERT INTO bloodhound_product (prefix, name) |
| VALUES ('p1', 'Product 1')""") |
| db("""INSERT INTO ticket (id, product) |
| VALUES (1, 'Product 1')""") |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| with self.product('p1'): |
| Ticket(self.env, 1) |
| |
| def test_can_upgrade_multi_product_from_v2(self): |
| mp = MultiProductSystem(self.env) |
| with self.env.db_direct_transaction as db: |
| mp._add_column_product_to_ticket(db) |
| mp._create_multiproduct_tables(db) |
| mp._replace_product_on_ticket_with_product_prefix(db) |
| mp._update_db_version(db, 2) |
| |
| db("""INSERT INTO bloodhound_product (prefix, name) |
| VALUES ('p1', 'Product 1')""") |
| db("""INSERT INTO ticket (id, product) |
| VALUES (1, 'p1')""") |
| db("""INSERT INTO ticket (id) |
| VALUES (2)""") |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| with self.product('p1'): |
| Ticket(self.env, 1) |
| with self.product('@'): |
| Ticket(self.env, 2) |
| |
| def test_upgrade_plugin(self): |
| self._enable_component(DummyPlugin) |
| self.env.upgrade() |
| |
| with self.env.db_direct_transaction as db: |
| db("SELECT v1 FROM dummy_table") |
| with self.assertFailsWithMissingColumn(): |
| db("SELECT v2 FROM dummy_table") |
| |
| DummyPlugin.version = 2 |
| self.env.upgrade() |
| |
| with self.env.db_direct_transaction as db: |
| db("SELECT v2 FROM dummy_table") |
| |
| def test_upgrade_plugin_to_multiproduct(self): |
| self._enable_multiproduct() |
| self._enable_component(DummyPlugin) |
| self.env.upgrade() |
| |
| with self.env.db_direct_transaction as db: |
| db("SELECT * FROM dummy_table") |
| db("""SELECT * FROM "@_dummy_table" """) |
| |
| def test_upgrade_existing_plugin_to_multiproduct(self): |
| self._enable_component(DummyPlugin) |
| self.env.upgrade() |
| with self.env.db_direct_transaction as db: |
| with self.assertFailsWithMissingTable(): |
| db("""SELECT * FROM "@_dummy_table" """) |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| with self.env.db_direct_transaction as db: |
| db("SELECT * FROM dummy_table") |
| db("""SELECT * FROM "@_dummy_table" """) |
| |
| def test_upgrading_existing_plugin_leaves_data_in_global_env(self): |
| DummyPlugin.version = 2 |
| self._enable_component(DummyPlugin) |
| self.env.upgrade() |
| with self.env.db_direct_transaction as db: |
| for i in range(5): |
| db("INSERT INTO dummy_table (v1) VALUES ('%d')" % i) |
| self.assertEqual( |
| len(db("SELECT * FROM dummy_table")), 5) |
| |
| self._enable_multiproduct() |
| self.env.upgrade() |
| with self.env.db_direct_transaction as db: |
| self.assertEqual( |
| len(db('SELECT * FROM "dummy_table"')), 5) |
| self.assertEqual( |
| len(db('SELECT * FROM "@_dummy_table"')), 0) |
| |
| def test_creating_new_product_calls_environment_created(self): |
| self._enable_component(DummyPlugin) |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| prod = Product(self.env) |
| prod.update_field_dict(dict(prefix='p1')) |
| ProductEnvironment(self.env, prod, create=True) |
| with self.env.db_direct_transaction as db: |
| db('SELECT * FROM "p1_dummy_table"') |
| |
| def test_migrating_to_multiproduct_with_custom_default_prefix(self): |
| ticket = self.insert_ticket('ticket') |
| |
| self.env.config.set('multiproduct', 'default_product_prefix', 'xxx') |
| self._enable_multiproduct() |
| self.env.upgrade() |
| |
| products = Product.select(self.env) |
| self.assertEqual(len(products), 1) |
| self.assertEqual(products[0].prefix, 'xxx') |
| |
| def _enable_multiproduct(self): |
| self._update_config('components', 'multiproduct.*', 'enabled') |
| |
| def _add_custom_field(self, field_name): |
| self._update_config('ticket-custom', field_name, 'text') |
| |
| def _enable_component(self, cls): |
| self._update_config( |
| 'components', |
| '%s.%s' % (cls.__module__, cls.__name__), |
| 'enabled' |
| ) |
| |
| def _update_config(self, section, key, value): |
| self.env.config.set(section, key, value) |
| self.env.config.save() |
| self.env = Environment(self.env_path) |
| |
| def _create_file_with_content(self, content): |
| filename = str(uuid.uuid4())[:6] |
| path = os.path.join(self.env_path, filename) |
| with open(path, 'wb') as f: |
| f.write(content) |
| return path |
| |
| @contextmanager |
| def assertFailsWithMissingTable(self): |
| with self.assertRaises(OperationalError) as cm: |
| yield |
| self.assertIn('no such table', str(cm.exception)) |
| |
| @contextmanager |
| def assertFailsWithMissingColumn(self): |
| with self.assertRaises(OperationalError) as cm: |
| yield |
| self.assertIn('no such column', str(cm.exception)) |
| |
| def create_ticket(self, summary, **kw): |
| ticket = Ticket(self.env) |
| ticket["summary"] = summary |
| for k, v in kw.items(): |
| ticket[k] = v |
| return ticket |
| |
| def insert_ticket(self, summary, **kw): |
| """Helper for inserting a ticket into the database""" |
| ticket = self.create_ticket(summary, **kw) |
| ticket.insert() |
| return ticket |
| |
| def create_wiki(self, name, text, **kw): |
| page = WikiPage(self.env, name) |
| page.text = text |
| for k, v in kw.items(): |
| page[k] = v |
| return page |
| |
| def insert_wiki(self, name, text = None, **kw): |
| text = text or "Dummy text" |
| page = self.create_wiki(name, text, **kw) |
| page.save("dummy author", "dummy comment", "::1") |
| return page |
| |
| def add_attachment(self, resource, path): |
| resource = '%s:%s' % (resource.realm, resource.id) |
| AttachmentAdmin(self.env)._do_add(resource, path) |
| |
| @contextmanager |
| def product(self, prefix): |
| old_env = self.env |
| self.env = ProductEnvironment(self.env, prefix) |
| yield |
| self.env = old_env |
| |
| |
| class DummyPlugin(Component): |
| implements(IEnvironmentSetupParticipant) |
| version = 1 |
| |
| def environment_created(self): |
| with self.env.db_transaction as db: |
| self.upgrade_environment(db) |
| |
| def environment_needs_upgrade(self, db): |
| return self.get_version(db) < self.version |
| |
| def upgrade_environment(self, db): |
| old_version = current_version = self.get_version(db) |
| db_connector, dummy = DatabaseManager(self.env)._get_connector() |
| |
| while current_version < self.version: |
| if current_version > 0: |
| db("CREATE TEMPORARY TABLE dummy_table_old AS " |
| "SELECT * FROM dummy_table") |
| db("DROP TABLE dummy_table") |
| |
| table = self.construct_dummy_table(current_version+1) |
| for statement in db_connector.to_sql(table): |
| db(statement) |
| |
| if current_version > 0: |
| cols = ['id'] + ['v%i' % (i+1) |
| for i in range(current_version+1)] |
| db("""INSERT INTO dummy_table (%s) |
| SELECT %s, '' FROM dummy_table_old |
| """ % (', '.join(cols), ', '.join(cols[:-1]))) |
| db("DROP TABLE dummy_table_old") |
| |
| current_version += 1 |
| |
| if current_version != old_version: |
| self.update_version(db, current_version) |
| |
| def construct_dummy_table(self, n_custom_fields=1): |
| fields = [Column('id')] + [ |
| Column('v%d' % (i+1)) for i in range(n_custom_fields) |
| ] |
| return Table('dummy_table')[fields] |
| |
| def get_version(self, db): |
| rows = db("SELECT value FROM system WHERE name = %s", |
| (self.__class__.__name__,)) |
| return int(rows[0][0]) if rows else 0 |
| |
| def update_version(self, db, version): |
| old_version = self.get_version(db) |
| if old_version: |
| db("UPDATE system SET value=%s WHERE name=%s", |
| (version, self.__class__.__name__,)) |
| else: |
| db("INSERT INTO system (name, value) VALUES ('%s','%s')" |
| % (self.__class__.__name__, version)) |
| return version |
| |
| |
| def test_suite(): |
| suite = unittest.TestSuite() |
| suite.addTest(unittest.makeSuite(EnvironmentUpgradeTestCase, 'test')) |
| return suite |
| |
| if __name__ == '__main__': |
| unittest.main(defaultTest='test_suite') |
