Python cache.CachedRepository类代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Python中trac.versioncontrol.cache.CachedRepository类的典型用法代码示例。如果您正苦于以下问题：Python CachedRepository类的具体用法？Python CachedRepository怎么用？Python CachedRepository使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

在下文中一共展示了CachedRepository类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_initial_sync

    def test_initial_sync(self):
        t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
        t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
        repos = self.get_repos(get_changeset=lambda x: changesets[int(x)],
                               youngest_rev=1)
        changes = [('trunk', Node.DIRECTORY, Changeset.ADD, None, None),
                   ('trunk/README', Node.FILE, Changeset.ADD, None, None)]
        changesets = [Mock(Changeset, repos, 0, '', '', t1,
                           get_changes=lambda: []),
                      Mock(Changeset, repos, 1, 'Import', 'joe', t2,
                           get_changes=lambda: iter(changes))]
        cache = CachedRepository(self.env, repos, self.log)
        cache.sync()

        cursor = self.db.cursor()
        cursor.execute("SELECT rev,time,author,message FROM revision")
        self.assertEquals(('0', to_utimestamp(t1), '', ''),
                          cursor.fetchone())
        self.assertEquals(('1', to_utimestamp(t2), 'joe', 'Import'),
                          cursor.fetchone())
        self.assertEquals(None, cursor.fetchone())
        cursor.execute("""
            SELECT rev,path,node_type,change_type,base_path,base_rev
            FROM node_change
            """)
        self.assertEquals(('1', 'trunk', 'D', 'A', None, None),
                          cursor.fetchone())
        self.assertEquals(('1', 'trunk/README', 'F', 'A', None, None),
                          cursor.fetchone())
        self.assertEquals(None, cursor.fetchone())

开发者ID:wiraqutra，项目名称:photrackjp，代码行数:30，代码来源:cache.py

示例2: sync

    def sync(self):
        youngest_stored = self.repos.get_youngest_rev_in_cache(self.db)
        if youngest_stored != str(self.repos.youngest_rev):
            # Need to cache all information for changes since the last
            # sync operation.

            if youngest_stored is None:
                youngest_stored = '0'

            # Obtain a list of changes since the last cache sync
            from p4trac.repos import _P4ChangesOutputConsumer
            output = _P4ChangesOutputConsumer(self.repos._repos)
            self.repos._connection.run('changes', '-l', '-s', 'submitted',
                                       '@>%s' % youngest_stored,
                                       output=output)

            if output.errors:
                from p4trac.repos import PerforceError
                raise PerforceError(output.errors)

            changes = output.changes
            changes.reverse()

            # Perform the precaching of the file history for files in these
            # changes.
            self.repos._repos.precacheFileHistoryForChanges(changes)

        # Call on to the default implementation now that we've cached
        # enough information to make it run a bit faster.
        CachedRepository.sync(self)

开发者ID:nyuhuhuu，项目名称:trachacks，代码行数:30，代码来源:api.py

示例3: test_get_changes

    def test_get_changes(self):
        t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
        t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
        cursor = self.db.cursor()
        cursor.execute("INSERT INTO revision (rev,time,author,message) "
                       "VALUES (0,%s,'','')", (to_timestamp(t1),))
        cursor.execute("INSERT INTO revision (rev,time,author,message) "
                       "VALUES (1,%s,'joe','Import')", (to_timestamp(t2),))
        cursor.executemany("INSERT INTO node_change (rev,path,node_type,"
                           "change_type,base_path,base_rev) "
                           "VALUES ('1',%s,%s,%s,%s,%s)",
                           [('trunk', 'D', 'A', None, None),
                            ('trunk/README', 'F', 'A', None, None)])
        cursor.execute("UPDATE system SET value='1' WHERE name='youngest_rev'")

        repos = Mock(Repository, 'test-repos', None, self.log,
                     get_changeset=lambda x: None,
                     get_youngest_rev=lambda: 1,
                     get_oldest_rev=lambda: 0,
                     next_rev=lambda x: None,
                     normalize_rev=lambda rev: rev)
        cache = CachedRepository(self.db, repos, None, self.log)
        self.assertEqual('1', cache.youngest_rev)
        changeset = cache.get_changeset(1)
        self.assertEqual('joe', changeset.author)
        self.assertEqual('Import', changeset.message)
        self.assertEqual(t2, changeset.date)
        changes = changeset.get_changes()
        self.assertEqual(('trunk', Node.DIRECTORY, Changeset.ADD, None, None),
                         changes.next())
        self.assertEqual(('trunk/README', Node.FILE, Changeset.ADD, None, None),
                         changes.next())
        self.assertRaises(StopIteration, changes.next)

开发者ID:gdgkyoto，项目名称:kyoto-gtug，代码行数:33，代码来源:cache.py

示例4: test_initial_sync

    def test_initial_sync(self):
        t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
        t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
        repos = self.get_repos(get_changeset=lambda x: changesets[int(x)], youngest_rev=1)
        changes = [
            ("trunk", Node.DIRECTORY, Changeset.ADD, None, None),
            ("trunk/README", Node.FILE, Changeset.ADD, None, None),
        ]
        changesets = [
            Mock(Changeset, repos, 0, "", "", t1, get_changes=lambda: []),
            Mock(Changeset, repos, 1, "Import", "joe", t2, get_changes=lambda: iter(changes)),
        ]
        cache = CachedRepository(self.env, repos, self.log)
        cache.sync()

        with self.env.db_query as db:
            rows = db("SELECT rev, time, author, message FROM revision")
            self.assertEquals(len(rows), 2)
            self.assertEquals(("0", to_utimestamp(t1), "", ""), rows[0])
            self.assertEquals(("1", to_utimestamp(t2), "joe", "Import"), rows[1])
            rows = db(
                """
                SELECT rev, path, node_type, change_type, base_path, base_rev
                FROM node_change"""
            )
            self.assertEquals(len(rows), 2)
            self.assertEquals(("1", "trunk", "D", "A", None, None), rows[0])
            self.assertEquals(("1", "trunk/README", "F", "A", None, None), rows[1])

开发者ID:rvelezc，项目名称:rafaelvelez.us-backup，代码行数:28，代码来源:cache.py

示例5: test_initial_sync

    def test_initial_sync(self):
        t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
        t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
        changes = [('trunk', Node.DIRECTORY, Changeset.ADD, None, None),
                   ('trunk/README', Node.FILE, Changeset.ADD, None, None)]
        changesets = [Mock(Changeset, 0, '', '', t1,
                           get_changes=lambda: []),
                      Mock(Changeset, 1, 'Import', 'joe', t2,
                           get_changes=lambda: iter(changes))]
        repos = Mock(Repository, 'test-repos', None, self.log,
                     get_changeset=lambda x: changesets[int(x)],
                     get_oldest_rev=lambda: 0,
                     get_youngest_rev=lambda: 1,
                     normalize_rev=lambda x: x,
                     next_rev=lambda x: int(x) == 0 and 1 or None)
        cache = CachedRepository(self.db, repos, None, self.log)
        cache.sync()

        cursor = self.db.cursor()
        cursor.execute("SELECT rev,time,author,message FROM revision")
        self.assertEquals(('0', to_timestamp(t1), '', ''), cursor.fetchone())
        self.assertEquals(('1', to_timestamp(t2), 'joe', 'Import'), cursor.fetchone())
        self.assertEquals(None, cursor.fetchone())
        cursor.execute("SELECT rev,path,node_type,change_type,base_path,"
                       "base_rev FROM node_change")
        self.assertEquals(('1', 'trunk', 'D', 'A', None, None),
                          cursor.fetchone())
        self.assertEquals(('1', 'trunk/README', 'F', 'A', None, None),
                          cursor.fetchone())
        self.assertEquals(None, cursor.fetchone())

开发者ID:gdgkyoto，项目名称:kyoto-gtug，代码行数:30，代码来源:cache.py

示例6: test_update_sync

    def test_update_sync(self):
        t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
        t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
        t3 = datetime(2003, 1, 1, 1, 1, 1, 0, utc)
        self.preset_cache(
            (("0", to_utimestamp(t1), "", ""), []),
            (
                ("1", to_utimestamp(t2), "joe", "Import"),
                [("trunk", "D", "A", None, None), ("trunk/README", "F", "A", None, None)],
            ),
        )
        repos = self.get_repos(get_changeset=lambda x: changesets[int(x)], youngest_rev=2)
        changes = [("trunk/README", Node.FILE, Changeset.EDIT, "trunk/README", 1)]
        changesets = [
            None,
            Mock(Changeset, repos, 1, "", "", t2, get_changes=lambda: []),
            Mock(Changeset, repos, 2, "Update", "joe", t3, get_changes=lambda: iter(changes)),
        ]
        cache = CachedRepository(self.env, repos, self.log)
        cache.sync()

        with self.env.db_query as db:
            self.assertEquals(
                [(to_utimestamp(t3), "joe", "Update")], db("SELECT time, author, message FROM revision WHERE rev='2'")
            )
            self.assertEquals(
                [("trunk/README", "F", "E", "trunk/README", "1")],
                db(
                    """SELECT path, node_type, change_type, base_path,
                                 base_rev
                          FROM node_change WHERE rev='2'"""
                ),
            )

开发者ID:rvelezc，项目名称:rafaelvelez.us-backup，代码行数:33，代码来源:cache.py

示例7: test_sync_changeset

    def test_sync_changeset(self):
        t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
        t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
        self.preset_cache(
            (('0', to_utimestamp(t1), '', ''), []),
            (('1', to_utimestamp(t2), 'joe', 'Import'),
             [('trunk', 'D', 'A', None, None),
              ('trunk/README', 'F', 'A', None, None)]),
            )
        repos = self.get_repos(get_changeset=lambda x: changesets[int(x)],
                               youngest_rev=1)
        changes1 = [('trunk', Node.DIRECTORY, Changeset.ADD, None, None),
                    ('trunk/README', Node.FILE, Changeset.ADD, None, None)]
        changesets = [
            Mock(Changeset, repos, 0, '**empty**', 'joe', t1,
                 get_changes=lambda: []),
            Mock(Changeset, repos, 1, 'Initial Import', 'joe', t2,
                 get_changes=lambda: iter(changes1)),
            ]
        cache = CachedRepository(self.env, repos, self.log)
        cache.sync_changeset(0)

        cursor = self.db.cursor()
        cursor.execute("SELECT time,author,message FROM revision ORDER BY rev")
        self.assertEquals((to_utimestamp(t1), 'joe', '**empty**'),
                          cursor.fetchone())
        self.assertEquals((to_utimestamp(t2), 'joe', 'Import'),
                          cursor.fetchone())
        self.assertEquals(None, cursor.fetchone())

开发者ID:wiraqutra，项目名称:photrackjp，代码行数:29，代码来源:cache.py

示例8: test_sync_changeset

    def test_sync_changeset(self):
        t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
        t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
        self.preset_cache(
            (("0", to_utimestamp(t1), "", ""), []),
            (
                ("1", to_utimestamp(t2), "joe", "Import"),
                [("trunk", "D", "A", None, None), ("trunk/README", "F", "A", None, None)],
            ),
        )
        repos = self.get_repos(get_changeset=lambda x: changesets[int(x)], youngest_rev=1)
        changes1 = [
            ("trunk", Node.DIRECTORY, Changeset.ADD, None, None),
            ("trunk/README", Node.FILE, Changeset.ADD, None, None),
        ]
        changesets = [
            Mock(Changeset, repos, 0, "**empty**", "joe", t1, get_changes=lambda: []),
            Mock(Changeset, repos, 1, "Initial Import", "joe", t2, get_changes=lambda: iter(changes1)),
        ]
        cache = CachedRepository(self.env, repos, self.log)
        cache.sync_changeset(0)

        rows = self.env.db_query("SELECT time, author, message FROM revision ORDER BY rev")
        self.assertEquals(2, len(rows))
        self.assertEquals((to_utimestamp(t1), "joe", "**empty**"), rows[0])
        self.assertEquals((to_utimestamp(t2), "joe", "Import"), rows[1])

开发者ID:rvelezc，项目名称:rafaelvelez.us-backup，代码行数:26，代码来源:cache.py

示例9: test_update_sync

    def test_update_sync(self):
        t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
        t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
        t3 = datetime(2003, 1, 1, 1, 1, 1, 0, utc)
        self.preset_cache(
            (('0', to_utimestamp(t1), '', ''), []),
            (('1', to_utimestamp(t2), 'joe', 'Import'),
             [('trunk', 'D', 'A', None, None),
              ('trunk/README', 'F', 'A', None, None)]),
            )
        repos = self.get_repos(get_changeset=lambda x: changesets[int(x)],
                               youngest_rev=2)
        changes = [('trunk/README', Node.FILE, Changeset.EDIT, 'trunk/README',
                    1)]
        changesets = [
            None,
            Mock(Changeset, repos, 1, '', '', t2, get_changes=lambda: []),
            Mock(Changeset, repos, 2, 'Update', 'joe', t3,
                 get_changes=lambda: iter(changes))
            ]
        cache = CachedRepository(self.env, repos, self.log)
        cache.sync()

        with self.env.db_query as db:
            self.assertEquals([(to_utimestamp(t3), 'joe', 'Update')],
                db("SELECT time, author, message FROM revision WHERE rev='2'"))
            self.assertEquals([('trunk/README', 'F', 'E', 'trunk/README',
                                '1')],
                    db("""SELECT path, node_type, change_type, base_path,
                                 base_rev
                          FROM node_change WHERE rev='2'"""))

开发者ID:thimalk，项目名称:bloodhound，代码行数:31，代码来源:cache.py

示例10: test_update_sync

    def test_update_sync(self):
        cursor = self.db.cursor()
        cursor.execute("INSERT INTO revision (rev,time,author,message) "
                       "VALUES (0,41000,'','')")
        cursor.execute("INSERT INTO revision (rev,time,author,message) "
                       "VALUES (1,42000,'joe','Import')")
        cursor.executemany("INSERT INTO node_change (rev,path,node_type,"
                           "change_type,base_path,base_rev) "
                           "VALUES ('1',%s,%s,%s,%s,%s)",
                           [('trunk', 'D', 'A', None, None),
                            ('trunk/README', 'F', 'A', None, None)])

        changes = [('trunk/README', Node.FILE, Changeset.EDIT, 'trunk/README', 1)]
        changeset = Mock(Changeset, 2, 'Update', 'joe', 42042,
                         get_changes=lambda: iter(changes))
        repos = Mock(Repository, 'test-repos', None, self.log,
                     get_changeset=lambda x: changeset,
                     get_youngest_rev=lambda: 2,
                     next_rev=lambda x: int(x) == 1 and 2 or None)
        cache = CachedRepository(self.db, repos, None, self.log)
        cache.sync()

        cursor = self.db.cursor()
        cursor.execute("SELECT time,author,message FROM revision WHERE rev='2'")
        self.assertEquals((42042, 'joe', 'Update'), cursor.fetchone())
        self.assertEquals(None, cursor.fetchone())
        cursor.execute("SELECT path,node_type,change_type,base_path,base_rev "
                       "FROM node_change WHERE rev='2'")
        self.assertEquals(('trunk/README', 'F', 'E', 'trunk/README', '1'),
                          cursor.fetchone())
        self.assertEquals(None, cursor.fetchone())

开发者ID:nyuhuhuu，项目名称:trachacks，代码行数:31，代码来源:cache.py

示例11: test_get_changes

    def test_get_changes(self):
        cursor = self.db.cursor()
        cursor.execute("INSERT INTO revision (rev,time,author,message) "
                       "VALUES (0,41000,'','')")
        cursor.execute("INSERT INTO revision (rev,time,author,message) "
                       "VALUES (1,42000,'joe','Import')")
        cursor.executemany("INSERT INTO node_change (rev,path,node_type,"
                           "change_type,base_path,base_rev) "
                           "VALUES ('1',%s,%s,%s,%s,%s)",
                           [('trunk', 'D', 'A', None, None),
                            ('trunk/README', 'F', 'A', None, None)])

        repos = Mock(Repository, 'test-repos', None, self.log,
                     get_changeset=lambda x: None,
                     get_youngest_rev=lambda: 1,
                     next_rev=lambda x: None, normalize_rev=lambda rev: rev)
        cache = CachedRepository(self.db, repos, None, self.log)
        self.assertEqual('1', cache.youngest_rev)
        changeset = cache.get_changeset(1)
        self.assertEqual('joe', changeset.author)
        self.assertEqual('Import', changeset.message)
        self.assertEqual(42000, changeset.date)
        changes = changeset.get_changes()
        self.assertEqual(('trunk', Node.DIRECTORY, Changeset.ADD, None, None),
                         changes.next())
        self.assertEqual(('trunk/README', Node.FILE, Changeset.ADD, None, None),
                         changes.next())
        self.assertRaises(StopIteration, changes.next)

开发者ID:nyuhuhuu，项目名称:trachacks，代码行数:28，代码来源:cache.py

示例12: test_initial_sync_with_empty_repos

    def test_initial_sync_with_empty_repos(self):
        repos = self.get_repos()
        cache = CachedRepository(self.env, repos, self.log)
        cache.sync()

        with self.env.db_query as db:
            self.assertEquals([], db("SELECT rev, time, author, message FROM revision"))
            self.assertEquals(0, db("SELECT COUNT(*) FROM node_change")[0][0])

开发者ID:rvelezc，项目名称:rafaelvelez.us-backup，代码行数:8，代码来源:cache.py

示例13: test_sync_changeset_if_not_exists

    def test_sync_changeset_if_not_exists(self):
        t = [
            datetime(2001, 1, 1, 1, 1, 1, 0, utc),  # r0
            datetime(2002, 1, 1, 1, 1, 1, 0, utc),  # r1
            datetime(2003, 1, 1, 1, 1, 1, 0, utc),  # r2
            datetime(2004, 1, 1, 1, 1, 1, 0, utc),  # r3
        ]
        self.preset_cache(
            (("0", to_utimestamp(t[0]), "joe", "**empty**"), []),
            (
                ("1", to_utimestamp(t[1]), "joe", "Import"),
                [("trunk", "D", "A", None, None), ("trunk/README", "F", "A", None, None)],
            ),
            # not exists r2
            (("3", to_utimestamp(t[3]), "joe", "Add COPYING"), [("trunk/COPYING", "F", "A", None, None)]),
        )
        repos = self.get_repos(get_changeset=lambda x: changesets[int(x)], youngest_rev=3)
        changes = [
            None,  # r0
            [
                ("trunk", Node.DIRECTORY, Changeset.ADD, None, None),  # r1
                ("trunk/README", Node.FILE, Changeset.ADD, None, None),
            ],
            [
                ("branches", Node.DIRECTORY, Changeset.ADD, None, None),  # r2
                ("tags", Node.DIRECTORY, Changeset.ADD, None, None),
            ],
            [("trunk/COPYING", Node.FILE, Changeset.ADD, None, None)],  # r3
        ]
        changesets = [
            Mock(Changeset, repos, 0, "**empty**", "joe", t[0], get_changes=lambda: []),
            Mock(Changeset, repos, 1, "Initial Import", "joe", t[1], get_changes=lambda: iter(changes[1])),
            Mock(Changeset, repos, 2, "Created directories", "john", t[2], get_changes=lambda: iter(changes[2])),
            Mock(Changeset, repos, 3, "Add COPYING", "joe", t[3], get_changes=lambda: iter(changes[3])),
        ]
        cache = CachedRepository(self.env, repos, self.log)
        self.assertRaises(NoSuchChangeset, cache.get_changeset, 2)
        cache.sync()
        self.assertRaises(NoSuchChangeset, cache.get_changeset, 2)

        self.assertEqual(None, cache.sync_changeset(2))
        cset = cache.get_changeset(2)
        self.assertEqual("john", cset.author)
        self.assertEqual("Created directories", cset.message)
        self.assertEqual(t[2], cset.date)
        cset_changes = cset.get_changes()
        self.assertEqual(("branches", Node.DIRECTORY, Changeset.ADD, None, None), cset_changes.next())
        self.assertEqual(("tags", Node.DIRECTORY, Changeset.ADD, None, None), cset_changes.next())
        self.assertRaises(StopIteration, cset_changes.next)

        rows = self.env.db_query("SELECT time,author,message FROM revision ORDER BY rev")
        self.assertEquals(4, len(rows))
        self.assertEquals((to_utimestamp(t[0]), "joe", "**empty**"), rows[0])
        self.assertEquals((to_utimestamp(t[1]), "joe", "Import"), rows[1])
        self.assertEquals((to_utimestamp(t[2]), "john", "Created directories"), rows[2])
        self.assertEquals((to_utimestamp(t[3]), "joe", "Add COPYING"), rows[3])

开发者ID:rvelezc，项目名称:rafaelvelez.us-backup，代码行数:56，代码来源:cache.py

示例14: test_initial_sync_with_empty_repos

    def test_initial_sync_with_empty_repos(self):
        repos = self.get_repos()
        cache = CachedRepository(self.env, repos, self.log)
        cache.sync()

        cursor = self.db.cursor()
        cursor.execute("SELECT rev,time,author,message FROM revision")
        self.assertEquals(None, cursor.fetchone())
        cursor.execute("SELECT COUNT(*) FROM node_change")
        self.assertEquals(0, cursor.fetchone()[0])

开发者ID:wiraqutra，项目名称:photrackjp，代码行数:10，代码来源:cache.py

示例15: test_clean_sync

    def test_clean_sync(self):
        t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
        t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
        t3 = datetime(2003, 1, 1, 1, 1, 1, 0, utc)
        self.preset_cache(
            (('0', to_utimestamp(t1), '', ''), []),
            (('1', to_utimestamp(t2), 'joe', 'Import'),
             [('trunk', 'D', 'A', None, None),
              ('trunk/README', 'F', 'A', None, None)]),
            )
        repos = self.get_repos(get_changeset=lambda x: changesets[int(x)],
                               youngest_rev=2)
        changes1 = [('trunk', Node.DIRECTORY, Changeset.ADD, None, None),
                    ('trunk/README', Node.FILE, Changeset.ADD, None, None)]
        changes2 = [('trunk/README', Node.FILE, Changeset.EDIT, 'trunk/README',
                     1)]
        changesets = [
            Mock(Changeset, repos, 0, '**empty**', 'joe', t1,
                 get_changes=lambda: []),
            Mock(Changeset, repos, 1, 'Initial Import', 'joe', t2,
                 get_changes=lambda: iter(changes1)),
            Mock(Changeset, repos, 2, 'Update', 'joe', t3,
                 get_changes=lambda: iter(changes2))
            ]
        cache = CachedRepository(self.env, repos, self.log)
        cache.sync(clean=True)

        cursor = self.db.cursor()
        cursor.execute("SELECT time,author,message FROM revision")
        self.assertEquals((to_utimestamp(t1), 'joe', '**empty**'),
                          cursor.fetchone())
        self.assertEquals((to_utimestamp(t2), 'joe', 'Initial Import'),
                          cursor.fetchone())
        self.assertEquals((to_utimestamp(t3), 'joe', 'Update'),
                          cursor.fetchone())
        self.assertEquals(None, cursor.fetchone())
        cursor.execute("""
            SELECT rev,path,node_type,change_type,base_path,base_rev
            FROM node_change ORDER BY rev
            """)
        self.assertEquals(('1', 'trunk', 'D', 'A', None, None),
                          cursor.fetchone())
        self.assertEquals(('1', 'trunk/README', 'F', 'A', None, None),
                          cursor.fetchone())
        self.assertEquals(('2', 'trunk/README', 'F', 'E',
                           'trunk/README', '1'),
                          cursor.fetchone())
        self.assertEquals(None, cursor.fetchone())

开发者ID:wiraqutra，项目名称:photrackjp，代码行数:48，代码来源:cache.py

示例16: test_initial_sync_with_empty_repos

    def test_initial_sync_with_empty_repos(self):
        changeset = Mock(Changeset, 0, '', '', 42000,
                         get_changes=lambda: [])
        repos = Mock(Repository, 'test-repos', None, self.log,
                     get_changeset=lambda x: changeset,
                     get_oldest_rev=lambda: 0,
                     get_youngest_rev=lambda: 0,
                     normalize_rev=lambda x: x,
                     next_rev=lambda x: None)
        cache = CachedRepository(self.db, repos, None, self.log)
        cache.sync()

        cursor = self.db.cursor()
        cursor.execute("SELECT rev,time,author,message FROM revision")
        self.assertEquals(('0', 42000, '', ''), cursor.fetchone())
        cursor.execute("SELECT COUNT(*) FROM node_change")
        self.assertEquals(0, cursor.fetchone()[0])

开发者ID:nyuhuhuu，项目名称:trachacks，代码行数:17，代码来源:cache.py

示例17: test_initial_sync_with_empty_repos

    def test_initial_sync_with_empty_repos(self):
        def no_changeset(rev):
            raise NoSuchChangeset(rev)

        repos = Mock(Repository, 'test-repos', None, self.log,
                     get_changeset=no_changeset,
                     get_oldest_rev=lambda: 1,
                     get_youngest_rev=lambda: 0,
                     normalize_rev=no_changeset,
                     next_rev=lambda x: None)
        cache = CachedRepository(self.db, repos, None, self.log)
        cache.sync()

        cursor = self.db.cursor()
        cursor.execute("SELECT rev,time,author,message FROM revision")
        self.assertEquals(None, cursor.fetchone())
        cursor.execute("SELECT COUNT(*) FROM node_change")
        self.assertEquals(0, cursor.fetchone()[0])

开发者ID:cyphactor，项目名称:lifecyclemanager，代码行数:18，代码来源:cache.py

示例18: test_clean_sync

    def test_clean_sync(self):
        t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
        t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
        t3 = datetime(2003, 1, 1, 1, 1, 1, 0, utc)
        self.preset_cache(
            (("0", to_utimestamp(t1), "", ""), []),
            (
                ("1", to_utimestamp(t2), "joe", "Import"),
                [("trunk", "D", "A", None, None), ("trunk/README", "F", "A", None, None)],
            ),
        )
        repos = self.get_repos(get_changeset=lambda x: changesets[int(x)], youngest_rev=2)
        changes1 = [
            ("trunk", Node.DIRECTORY, Changeset.ADD, None, None),
            ("trunk/README", Node.FILE, Changeset.ADD, None, None),
        ]
        changes2 = [("trunk/README", Node.FILE, Changeset.EDIT, "trunk/README", 1)]
        changesets = [
            Mock(Changeset, repos, 0, "**empty**", "joe", t1, get_changes=lambda: []),
            Mock(Changeset, repos, 1, "Initial Import", "joe", t2, get_changes=lambda: iter(changes1)),
            Mock(Changeset, repos, 2, "Update", "joe", t3, get_changes=lambda: iter(changes2)),
        ]
        cache = CachedRepository(self.env, repos, self.log)
        cache.sync(clean=True)

        rows = self.env.db_query(
            """
            SELECT time, author, message FROM revision ORDER BY rev
            """
        )
        self.assertEquals(3, len(rows))
        self.assertEquals((to_utimestamp(t1), "joe", "**empty**"), rows[0])
        self.assertEquals((to_utimestamp(t2), "joe", "Initial Import"), rows[1])
        self.assertEquals((to_utimestamp(t3), "joe", "Update"), rows[2])

        rows = self.env.db_query(
            """
            SELECT rev, path, node_type, change_type, base_path, base_rev
            FROM node_change ORDER BY rev, path"""
        )
        self.assertEquals(3, len(rows))
        self.assertEquals(("1", "trunk", "D", "A", None, None), rows[0])
        self.assertEquals(("1", "trunk/README", "F", "A", None, None), rows[1])
        self.assertEquals(("2", "trunk/README", "F", "E", "trunk/README", "1"), rows[2])

开发者ID:rvelezc，项目名称:rafaelvelez.us-backup，代码行数:44，代码来源:cache.py

示例19: get_repository

    def get_repository(self, type, dir, authname):
        """Return a `SubversionRepository`.

        The repository is wrapped in a `CachedRepository`, unless `type` is
        'direct-svnfs'.
        """
        if not self._version:
            self._version = self._get_version()
            self.env.systeminfo.append(("Subversion", self._version))
        fs_repos = SubversionRepository(dir, None, self.log, {"tags": self.tags, "branches": self.branches})
        if type == "direct-svnfs":
            repos = fs_repos
        else:
            repos = CachedRepository(self.env.get_db_cnx, fs_repos, None, self.log)
            repos.has_linear_changesets = True
        if authname:
            authz = SubversionAuthorizer(self.env, weakref.proxy(repos), authname)
            repos.authz = fs_repos.authz = authz
        return repos

开发者ID:gdgkyoto，项目名称:kyoto-gtug，代码行数:19，代码来源:svn_fs.py

示例20: test_get_changes

 def test_get_changes(self):
     t1 = datetime(2001, 1, 1, 1, 1, 1, 0, utc)
     t2 = datetime(2002, 1, 1, 1, 1, 1, 0, utc)
     self.preset_cache(
         (("0", to_utimestamp(t1), "", ""), []),
         (
             ("1", to_utimestamp(t2), "joe", "Import"),
             [("trunk", "D", "A", None, None), ("trunk/RDME", "F", "A", None, None)],
         ),
     )
     repos = self.get_repos()
     cache = CachedRepository(self.env, repos, self.log)
     self.assertEqual("1", cache.youngest_rev)
     changeset = cache.get_changeset(1)
     self.assertEqual("joe", changeset.author)
     self.assertEqual("Import", changeset.message)
     self.assertEqual(t2, changeset.date)
     changes = changeset.get_changes()
     self.assertEqual(("trunk", Node.DIRECTORY, Changeset.ADD, None, None), changes.next())
     self.assertEqual(("trunk/RDME", Node.FILE, Changeset.ADD, None, None), changes.next())
     self.assertRaises(StopIteration, changes.next)

开发者ID:rvelezc，项目名称:rafaelvelez.us-backup，代码行数:21，代码来源:cache.py

注：本文中的trac.versioncontrol.cache.CachedRepository类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python api.Request类代码示例发布时间：2022-05-27

Python api.RepositoryManager类代码示例发布时间：2022-05-27

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13762|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：9354|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4061|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4039|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3816|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3499|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3022|2022-01-22

8 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2635|2022-05-25

9 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2622|2022-01-22

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2290|2022-01-22

客服电话

电子邮件

Python cache.CachedRepository类代码示例

示例1: test_initial_sync

示例2: sync

示例3: test_get_changes

示例4: test_initial_sync

示例5: test_initial_sync

示例6: test_update_sync

示例7: test_sync_changeset

示例8: test_sync_changeset

示例9: test_update_sync

示例10: test_update_sync

示例11: test_get_changes

示例12: test_initial_sync_with_empty_repos

示例13: test_sync_changeset_if_not_exists

示例14: test_initial_sync_with_empty_repos

示例15: test_clean_sync

示例16: test_initial_sync_with_empty_repos

示例17: test_initial_sync_with_empty_repos

示例18: test_clean_sync

示例19: get_repository

示例20: test_get_changes

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053