docker-utils/docker-diff at master · a-ba/docker-utils · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
#!/usr/bin/python3
# vim:et:nosta:sw=4:sts=4:

import argparse, difflib, subprocess, json, contextlib, signal, tempfile, os, sys, io

from subprocess import Popen, PIPE

def default_run_error_handler(cmd, code, out, err):
    sys.stderr.buffer.write(err)
    raise subprocess.CalledProcessError(code, cmd)

def run(cmd, *, error_handler=default_run_error_handler):
#    sys.stderr.write("run %s\n" % " ".join(map(repr, cmd)))
    proc = Popen(cmd, stdin=PIPE, stdout=PIPE, stderr=PIPE,
            preexec_fn = lambda: signal.signal(signal.SIGINT, signal.SIG_IGN))
    out, err = proc.communicate()
    code = proc.wait()
    if code:
        return error_handler(cmd, code, out, err)
    sys.stderr.buffer.write(err)
    return out.decode().strip()

def inspect(ctr):
    lst = json.loads(run(["docker", "inspect", ctr]))
    assert len(lst) == 1
    return lst[0]

class Deferrer:
    def __init__(self):
        self._depth  = 0
        self._caught = None

    def _handler(self, sig, frame):
        assert isinstance(self._caught, bool)
        self._caught = True

    def __enter__(self):
        assert self._depth >= 0
        if not self._depth:
            assert self._caught is None
            self._caught = False
            signal.signal(signal.SIGINT, self._handler)
        else:
            assert isinstance(self._caught, bool)
        self._depth += 1
        return self

    def __exit__(self, a, b, c):
        assert self._depth > 0
        assert isinstance(self._caught, bool)
        self._depth -= 1
        if not self._depth:
            caught = self._caught
            self._caught = None
            if caught:
                raise KeyboardInterrupt()
defer_sigint = Deferrer()


@contextlib.contextmanager
def run_tmp_container(image):
    ctr = None
    try:
        with defer_sigint:
            #TODO: use docker create
            ctr = run(["docker", "run", "-d", image, "/bin/true"])
        yield ctr
    finally:
        if ctr:
            with defer_sigint:
                run(["docker", "rm", "-f", ctr])


parser = argparse.ArgumentParser()

parser.add_argument("container", metavar="CONTAINER", action="store",
        help = "container to be diffed")

parser.add_argument("files", metavar="FILE", action="store",
        nargs="+", help = "files to be diffed")

args = parser.parse_args()

image = inspect(args.container)["Image"]

with run_tmp_container(image) as tmpctr, \
        tempfile.TemporaryDirectory() as tmpdir:
        orig = os.path.join(tmpdir, "orig")
        dest = os.path.join(tmpdir, "dest")
        for path in args.files:
            bn = os.path.basename(path)
            path = path.strip("/")
            assert bn, "must not be a directory"
            with defer_sigint:
                for cp_src, cp_dst in (
                    ("%s:/%s" % (tmpctr,         path), orig),
                    ("%s:/%s" % (args.container, path), dest)):

                    def hnd(cmd, code, out, err):
                        if b"no such file or directory" in err:
                            with open(cp_dst, "wb"):
                                pass
                        else:
                            default_run_error_handler(cmd, code, out, err)
                    run(["docker", "cp", cp_src, cp_dst], error_handler=hnd)
            with    open(orig, errors="surrogateescape") as a, \
                    open(dest, errors="surrogateescape") as b:
                print("".join(difflib.unified_diff(list(a), list(b), path, path)))