""" Migration script for old 'yaml' and 'json' cassettes .. warning:: Backup your cassettes files before migration. It merges and deletes the request obsolete keys (protocol, host, port, path) into new 'uri' key. Usage:: python -m vcr.migration PATH The PATH can be path to the directory with cassettes or cassette itself """ import json import os import re import shutil import sys import tempfile PARTS = [ 'protocol', 'host', 'port', 'path', ] def build_uri(**parts): return "{protocol}://{host}:{port}{path}".format(**parts) def migrate_json(in_fp, out_fp): data = json.load(in_fp) for item in data: req = item['request'] uri = dict((k, req.pop(k)) for k in PARTS) req['uri'] = build_uri(**uri) json.dump(data, out_fp, indent=4) def migrate_yml(in_fp, out_fp): migrated = False uri = dict.fromkeys(PARTS, None) for line in in_fp: for part in uri: match = re.match('\s+{0}:\s(.*)'.format(part), line) if match: uri[part] = match.group(1) break else: out_fp.write(line) if None not in uri.values(): # if all uri parts are collected out_fp.write(" uri: {0}\n".format(build_uri(**uri))) uri = dict.fromkeys(PARTS, None) # reset dict migrated = True if not migrated: raise RuntimeError("migration failed") def migrate(file_path, migration_fn): # because we assume that original files can be reverted # we will try to copy the content. (os.rename not needed) with tempfile.TemporaryFile(mode='w+') as out_fp: with open(file_path, 'r') as in_fp: migration_fn(in_fp, out_fp) with open(file_path, 'w') as in_fp: out_fp.seek(0) shutil.copyfileobj(out_fp, in_fp) def try_migrate(path): try: # try to migrate as json migrate(path, migrate_json) except Exception: # probably the file is not a json try: # let's try to migrate as yaml migrate(path, migrate_yml) except Exception: # oops probably the file is not a cassette return False return True def main(): if len(sys.argv) != 2: raise SystemExit("Please provide path to cassettes directory or file. " "Usage: python -m vcr.migration PATH") path = sys.argv[1] if not os.path.isabs(path): path = os.path.abspath(path) files = [path] if os.path.isdir(path): files = (os.path.join(root, name) for (root, dirs, files) in os.walk(path) for name in files) for file_path in files: migrated = try_migrate(file_path) status = 'OK' if migrated else 'FAIL' sys.stderr.write("[{0}] {1}\n".format(status, file_path)) sys.stderr.write("Done.\n") if __name__ == '__main__': main()