import unittest import actions import schema import table_data_set import migrations class TestMigrations(unittest.TestCase): def test_migrations(self): tdset = table_data_set.TableDataSet() tdset.apply_doc_actions(schema_version0()) migration_actions = migrations.create_migrations(tdset.all_tables) tdset.apply_doc_actions(migration_actions) # Compare schema derived from migrations to the current schema. migrated_schema = tdset.get_schema() current_schema = {a.table_id: {c['id']: c for c in a.columns} for a in schema.schema_create_actions()} # pylint: disable=too-many-nested-blocks if migrated_schema != current_schema: # Figure out the version of new migration to suggest, and whether to update SCHEMA_VERSION. new_version = max(schema.SCHEMA_VERSION, migrations.get_last_migration_version() + 1) # Figure out the missing actions. doc_actions = [] for table_id in sorted(current_schema.viewkeys() | migrated_schema.viewkeys()): if table_id not in migrated_schema: doc_actions.append(actions.AddTable(table_id, current_schema[table_id].values())) elif table_id not in current_schema: doc_actions.append(actions.RemoveTable(table_id)) else: current_cols = current_schema[table_id] migrated_cols = migrated_schema[table_id] for col_id in sorted(current_cols.viewkeys() | migrated_cols.viewkeys()): if col_id not in migrated_cols: doc_actions.append(actions.AddColumn(table_id, col_id, current_cols[col_id])) elif col_id not in current_cols: doc_actions.append(actions.RemoveColumn(table_id, col_id)) else: current_info = current_cols[col_id] migrated_info = migrated_cols[col_id] delta = {k: v for k, v in current_info.iteritems() if v != migrated_info.get(k)} if delta: doc_actions.append(actions.ModifyColumn(table_id, col_id, delta)) suggested_migration = ( "----------------------------------------------------------------------\n" + "*** migrations.py ***\n" + "----------------------------------------------------------------------\n" + "@migration(schema_version=%s)\n" % new_version + "def migration%s(tdset):\n" % new_version + " return tdset.apply_doc_actions([\n" + "".join(stringify(a) + ",\n" for a in doc_actions) + " ])\n" ) if new_version != schema.SCHEMA_VERSION: suggested_schema_update = ( "----------------------------------------------------------------------\n" + "*** schema.py ***\n" + "----------------------------------------------------------------------\n" + "SCHEMA_VERSION = %s\n" % new_version ) else: suggested_schema_update = "" self.fail("Migrations are incomplete. Suggested migration to add:\n" + suggested_schema_update + suggested_migration) def stringify(doc_action): if isinstance(doc_action, actions.AddColumn): return ' add_column(%r, %s)' % (doc_action.table_id, col_info_args(doc_action.col_info)) elif isinstance(doc_action, actions.AddTable): return (' actions.AddTable(%r, [\n' % doc_action.table_id + ''.join(' schema.make_column(%s),\n' % col_info_args(c) for c in doc_action.columns) + ' ])') else: return " actions.%s(%s)" % (doc_action.__class__.__name__, ", ".join(map(repr, doc_action))) def col_info_args(col_info): extra = "" for k in ("formula", "isFormula"): v = col_info.get(k) if v: extra += ", %s=%r" % (k, v) return "%r, %r%s" % (col_info['id'], col_info['type'], extra) def schema_version0(): # This is the initial version of the schema before the very first migration. It's a historical # snapshot, and thus should not be edited. The test verifies that starting with this v0, # migrations bring the schema to the current version. def make_column(col_id, col_type, formula='', isFormula=False): return { "id": col_id, "type": col_type, "isFormula": isFormula, "formula": formula } return [ actions.AddTable("_grist_DocInfo", [ make_column("docId", "Text"), make_column("peers", "Text"), make_column("schemaVersion", "Int"), ]), actions.AddTable("_grist_Tables", [ make_column("tableId", "Text"), ]), actions.AddTable("_grist_Tables_column", [ make_column("parentId", "Ref:_grist_Tables"), make_column("parentPos", "PositionNumber"), make_column("colId", "Text"), make_column("type", "Text"), make_column("widgetOptions","Text"), make_column("isFormula", "Bool"), make_column("formula", "Text"), make_column("label", "Text") ]), actions.AddTable("_grist_Imports", [ make_column("tableRef", "Ref:_grist_Tables"), make_column("origFileName", "Text"), make_column("parseFormula", "Text", isFormula=True, formula="grist.parseImport(rec, table._engine)"), make_column("delimiter", "Text", formula="','"), make_column("doublequote", "Bool", formula="True"), make_column("escapechar", "Text"), make_column("quotechar", "Text", formula="'\"'"), make_column("skipinitialspace", "Bool"), make_column("encoding", "Text", formula="'utf8'"), make_column("hasHeaders", "Bool"), ]), actions.AddTable("_grist_External_database", [ make_column("host", "Text"), make_column("port", "Int"), make_column("username", "Text"), make_column("dialect", "Text"), make_column("database", "Text"), make_column("storage", "Text"), ]), actions.AddTable("_grist_External_table", [ make_column("tableRef", "Ref:_grist_Tables"), make_column("databaseRef", "Ref:_grist_External_database"), make_column("tableName", "Text"), ]), actions.AddTable("_grist_TabItems", [ make_column("tableRef", "Ref:_grist_Tables"), make_column("viewRef", "Ref:_grist_Views"), ]), actions.AddTable("_grist_Views", [ make_column("name", "Text"), make_column("type", "Text"), make_column("layoutSpec", "Text"), ]), actions.AddTable("_grist_Views_section", [ make_column("tableRef", "Ref:_grist_Tables"), make_column("parentId", "Ref:_grist_Views"), make_column("parentKey", "Text"), make_column("title", "Text"), make_column("defaultWidth", "Int", formula="100"), make_column("borderWidth", "Int", formula="1"), make_column("theme", "Text"), make_column("chartType", "Text"), make_column("layoutSpec", "Text"), make_column("filterSpec", "Text"), make_column("sortColRefs", "Text"), make_column("linkSrcSectionRef", "Ref:_grist_Views_section"), make_column("linkSrcColRef", "Ref:_grist_Tables_column"), make_column("linkTargetColRef", "Ref:_grist_Tables_column"), ]), actions.AddTable("_grist_Views_section_field", [ make_column("parentId", "Ref:_grist_Views_section"), make_column("parentPos", "PositionNumber"), make_column("colRef", "Ref:_grist_Tables_column"), make_column("width", "Int"), make_column("widgetOptions","Text"), ]), actions.AddTable("_grist_Validations", [ make_column("formula", "Text"), make_column("name", "Text"), make_column("tableRef", "Int") ]), actions.AddTable("_grist_REPL_Hist", [ make_column("code", "Text"), make_column("outputText", "Text"), make_column("errorText", "Text") ]), actions.AddTable("_grist_Attachments", [ make_column("fileIdent", "Text"), make_column("fileName", "Text"), make_column("fileType", "Text"), make_column("fileSize", "Int"), make_column("timeUploaded", "DateTime") ]), actions.AddRecord("_grist_DocInfo", 1, {}) ] if __name__ == "__main__": unittest.main()