git-filter-repo/t/t9391/unusual.py

132 lines
4.5 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
# Please: DO NOT USE THIS AS AN EXAMPLE.
#
# This file is NOT for demonstration of how to use git-filter-repo as a
# libary; it exists to test corner cases or otherwise unusual inputs, and
# to verify some invariants that git-filter-repo currently aims to maintain
# (these invariants might be different in future versions of
# git-filter-repo). As such, it reaches deep into the internals and does
# weird things that you should probably avoid in your usage of
# git-filter-repo. Any code in this testcase is much more likely to have
# API breaks than other files in t9391.
import collections
import os
import random
import io
import sys
import textwrap
import git_filter_repo as fr
total_objects = {'common': 0, 'uncommon': 0}
def track_everything(obj, *_ignored):
if type(obj) == fr.Blob or type(obj) == fr.Commit:
total_objects['common'] += 1
else:
total_objects['uncommon'] += 1
if type(obj) == fr.Reset:
def assert_not_reached(x): raise SystemExit("should have been skipped!")
obj.dump = assert_not_reached
obj.skip()
if hasattr(obj, 'id') and type(obj) != fr.Tag:
# The creation of myblob should cause objects in stream to get their ids
# increased by 1; this shouldn't be depended upon as API by external
# projects, I'm just verifying an invariant of the current code.
assert fr._IDS._reverse_translation[obj.id] == [obj.id - 1]
def handle_progress(progress):
print(b"Decipher this: "+bytes(reversed(progress.message)))
track_everything(progress)
def handle_checkpoint(checkpoint_object):
# Flip a coin; see if we want to pass the checkpoint through.
if random.randint(0,1) == 0:
checkpoint_object.dump(parser._output)
track_everything(checkpoint_object)
mystr = b'This is the contents of the blob'
compare = b"Blob:\n blob\n mark :1\n data %d\n %s" % (len(mystr), mystr)
# Next line's only purpose is testing code coverage of something that helps
# debugging git-filter-repo; it is NOT something external folks should depend
# upon.
myblob = fr.Blob(mystr)
assert bytes(myblob) == compare
# Everyone should be using RepoFilter objects, not FastExportParser. But for
# testing purposes...
parser = fr.FastExportParser(blob_callback = track_everything,
reset_callback = track_everything,
commit_callback = track_everything,
tag_callback = track_everything,
progress_callback = handle_progress,
checkpoint_callback = handle_checkpoint)
parser.run(input = sys.stdin.detach(),
output = open(os.devnull, 'bw'))
# DO NOT depend upon or use _IDS directly you external script writers. I'm
# only testing here for code coverage; the capacity exists to help debug
# git-filter-repo itself, not for external folks to use.
assert str(fr._IDS).startswith("Current count: 5")
print("Found {} blobs/commits and {} other objects"
.format(total_objects['common'], total_objects['uncommon']))
stream = io.BytesIO(textwrap.dedent('''
blob
mark :1
data 5
hello
commit refs/heads/A
mark :2
author Just Me <just@here.org> 1234567890 -0200
committer Just Me <just@here.org> 1234567890 -0200
data 2
A
commit refs/heads/B
mark :3
author Just Me <just@here.org> 1234567890 -0200
committer Just Me <just@here.org> 1234567890 -0200
data 2
B
from :2
M 100644 :1 greeting
reset refs/heads/B
from :3
commit refs/heads/C
mark :4
author Just Me <just@here.org> 1234567890 -0200
committer Just Me <just@here.org> 1234567890 -0200
data 2
C
from :3
M 100644 :1 salutation
'''[1:]).encode())
counts = collections.Counter()
def look_for_reset(obj, metadata):
print("Processing {}".format(obj))
counts[type(obj)] += 1
if type(obj) == fr.Reset:
assert obj.ref == b'refs/heads/B'
# Use all kinds of internals that external scripts should NOT use and which
# are likely to break in the future, just to verify a few invariants...
args = fr.FilteringOptions.parse_args(['--stdin', '--dry-run',
'--path', 'salutation'])
filter = fr.RepoFilter(args,
blob_callback = look_for_reset,
reset_callback = look_for_reset,
commit_callback = look_for_reset,
tag_callback = look_for_reset)
filter._input = stream
filter._setup_output()
filter._sanity_checks_handled = True
filter.run()
assert counts == collections.Counter({fr.Blob: 1, fr.Commit: 3, fr.Reset: 1})