python: Fix version check.
[gpgme.git] / lang / python / tests / support.py
1 # Copyright (C) 2016 g10 Code GmbH
2 #
3 # This file is part of GPGME.
4 #
5 # GPGME is free software; you can redistribute it and/or modify it
6 # under the terms of the GNU General Public License as published by
7 # the Free Software Foundation; either version 2 of the License, or
8 # (at your option) any later version.
9 #
10 # GPGME is distributed in the hope that it will be useful, but WITHOUT
11 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
12 # or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General
13 # Public License for more details.
14 #
15 # You should have received a copy of the GNU Lesser General Public
16 # License along with this program; if not, see <http://www.gnu.org/licenses/>.
17
18 from __future__ import absolute_import, print_function, unicode_literals
19 del absolute_import, print_function, unicode_literals
20
21 import contextlib
22 import shutil
23 import sys
24 import os
25 import re
26 import tempfile
27 import time
28 import gpg
29
30 def assert_gpg_version(version=(2, 1, 0)):
31     with gpg.Context() as c:
32         clean_version = re.match(r'\d+\.\d+\.\d+', c.engine_info.version).group(0)
33         if tuple(map(int, clean_version.split('.'))) < version:
34             print("GnuPG too old: have {0}, need {1}.".format(
35                 c.engine_info.version, '.'.join(version)))
36             sys.exit(77)
37
38 # known keys
39 alpha = "A0FF4590BB6122EDEF6E3C542D727CC768697734"
40 bob = "D695676BDCEDCC2CDD6152BCFE180B1DA9E3B0B2"
41 encrypt_only = "F52770D5C4DB41408D918C9F920572769B9FE19C"
42 sign_only = "7CCA20CCDE5394CEE71C9F0BFED153F12F18F45D"
43 no_such_key = "A" * 40
44
45 def make_filename(name):
46     return os.path.join(os.environ['top_srcdir'], 'tests', 'gpg', name)
47
48 def in_srcdir(name):
49     return os.path.join(os.environ['srcdir'], name)
50
51 verbose = int(os.environ.get('verbose', 0)) > 1
52 def print_data(data):
53     if verbose:
54         try:
55             # See if it is a file-like object.
56             data.seek(0, os.SEEK_SET)
57             data = data.read()
58         except:
59             # Hope for the best.
60             pass
61
62         if hasattr(sys.stdout, "buffer"):
63             sys.stdout.buffer.write(data)
64         else:
65             sys.stdout.write(data)
66
67 def mark_key_trusted(ctx, key):
68     class Editor(object):
69         def __init__(self):
70             self.steps = ["trust", "save"]
71         def edit(self, status, args, out):
72             if args == "keyedit.prompt":
73                 result = self.steps.pop(0)
74             elif args == "edit_ownertrust.value":
75                 result = "5"
76             elif args == "edit_ownertrust.set_ultimate.okay":
77                 result = "Y"
78             elif args == "keyedit.save.okay":
79                 result = "Y"
80             else:
81                 result = None
82             return result
83     with gpg.Data() as sink:
84         ctx.op_edit(key, Editor().edit, sink, sink)
85
86
87 # Python3.2 and up has tempfile.TemporaryDirectory, but we cannot use
88 # that, because there shutil.rmtree is used without
89 # ignore_errors=True, and that races against gpg-agent deleting its
90 # sockets.
91 class TemporaryDirectory(object):
92     def __enter__(self):
93         self.path = tempfile.mkdtemp()
94         return self.path
95     def __exit__(self, *args):
96         shutil.rmtree(self.path, ignore_errors=True)
97
98 @contextlib.contextmanager
99 def EphemeralContext():
100     with TemporaryDirectory() as tmp:
101         home = os.environ['GNUPGHOME']
102         shutil.copy(os.path.join(home, "gpg.conf"), tmp)
103         shutil.copy(os.path.join(home, "gpg-agent.conf"), tmp)
104
105         with gpg.Context(home_dir=tmp) as ctx:
106             yield ctx
107
108             # Ask the agent to quit.
109             agent_socket = os.path.join(tmp, "S.gpg-agent")
110             ctx.protocol = gpg.constants.protocol.ASSUAN
111             ctx.set_engine_info(ctx.protocol, file_name=agent_socket)
112             ctx.assuan_transact(["KILLAGENT"])
113
114             # Block until it is really gone.
115             while os.path.exists(agent_socket):
116                 time.sleep(.01)