added monkey patching via fuse for full integration test

.travis.yml

@@ -1,3 +1,5 @@
+sudo: required
+dist: trusty
 language: python
 python:
   - "2.6"
@@ -6,8 +8,10 @@ python:
   - "3.4"
   - "3.5"
   - "nightly"
+before_install:
+  - sudo apt-get install fuse
 install:
-  - pip install coveralls
+  - pip install -r tests/requirements.txt
 script:
   - coverage run -m unittest tests
 after_success:

acme_tiny.py

@@ -118,8 +118,10 @@ def get_crt(account_key, csr, acme_dir, log=LOGGER, CA=DEFAULT_CA):
         wellknown_url = "http://{0}/.well-known/acme-challenge/{1}".format(domain, token)
         try:
             resp = urlopen(wellknown_url)
-            assert resp.read().decode('utf8').strip() == keyauthorization
+            resp_data = resp.read().decode('utf8').strip()
+            assert resp_data == keyauthorization
         except (IOError, AssertionError):
+            print("resp_data", resp_data)
             os.remove(wellknown_path)
             raise ValueError("Wrote file to {0}, but couldn't download {1}".format(
                 wellknown_path, wellknown_url))

tests/monkey.py

@@ -0,0 +1,53 @@
+import os, sys
+from tempfile import NamedTemporaryFile
+from subprocess import Popen
+from fuse import FUSE, Operations, LoggingMixIn
+try:
+    from urllib.request import urlopen # Python 3
+except ImportError:
+    from urllib2 import urlopen # Python 2
+
+# domain with server.py running on it for testing
+DOMAIN = os.getenv("TRAVIS_DOMAIN", "travis-ci.gethttpsforfree.com")
+
+# generate account and domain keys
+def gen_keys():
+    # good account key
+    account_key = NamedTemporaryFile()
+    Popen(["openssl", "genrsa", "-out", account_key.name, "2048"]).wait()
+
+    # good domain key
+    domain_key = NamedTemporaryFile()
+    domain_csr = NamedTemporaryFile()
+    Popen(["openssl", "req", "-newkey", "rsa:2048", "-nodes", "-keyout", domain_key.name,
+        "-subj", "/CN={0}".format(DOMAIN), "-out", domain_csr.name]).wait()
+    return {
+        "account_key": account_key,
+        "domain_key": domain_key,
+        "domain_csr": domain_csr,
+    }
+
+# fake a folder structure to catch the key authorization file
+FS = {}
+class Passthrough(LoggingMixIn, Operations):
+    def getattr(self, path, fh=None):
+        f = FS.get(path, None)
+        if f is None:
+            return super(Passthrough, self).getattr(path, fh=fh)
+        return f
+
+    def write(self, path, buf, offset, fh):
+        urlopen("http://{0}/.well-known/acme-challenge/?{1}".format(DOMAIN,
+            os.getenv("TRAVIS_SESSION", "not_set")), buf)
+        return len(buf)
+
+    def create(self, path, mode, fi=None):
+        FS[path] = {"st_mode": 33204}
+        return 0
+
+    def unlink(self, path):
+        del(FS[path])
+        return 0
+
+if __name__ == "__main__":
+    FUSE(Passthrough(), sys.argv[1], nothreads=True, foreground=True)

tests/requirements.txt

@@ -0,0 +1,2 @@
+coveralls
+fusepy

tests/server.py

@@ -0,0 +1,28 @@
+import os, re, hmac
+from wsgiref.simple_server import make_server
+
+KEY_AUTHORIZATION = {"uri": "/not_set", "data": ""}
+TRAVIS_SESSION = os.getenv("TRAVIS_SESSION", "not_yet_set")
+
+def app(req, resp):
+    if req['REQUEST_METHOD'] == "POST":
+        if hmac.compare_digest(req['QUERY_STRING'], TRAVIS_SESSION):
+            body_len = min(int(req['CONTENT_LENGTH']), 90)
+            body = req['wsgi.input'].read(body_len).decode("utf8")
+            body = re.sub(r"[^A-Za-z0-9_\-\.]", "_", body)
+            KEY_AUTHORIZATION['uri'] = "/{0}".format(body.split(".", 1)[0])
+            KEY_AUTHORIZATION['body'] = body
+            resp('201 Created', [])
+            return ["".encode("utf8")]
+        else:
+            resp("403 Forbidden", [])
+            return ["403 Forbidden".encode("utf8")]
+    else:
+        if hmac.compare_digest(req['PATH_INFO'], KEY_AUTHORIZATION['uri']):
+            resp('200 OK', [])
+            return [KEY_AUTHORIZATION['data'].encode("utf8")]
+        else:
+            resp("404 Not Found", [])
+            return ["404 Not Found".encode("utf8")]
+
+make_server("localhost", 8888, app).serve_forever()

tests/test_module.py

@@ -1,11 +1,26 @@
-import unittest
+import unittest, os, tempfile
+from subprocess import Popen
+
 import acme_tiny
+from . import monkey
 
 class TestModule(unittest.TestCase):
     "Tests for acme_tiny.get_crt()"
 
     def setUp(self):
         self.CA = "https://acme-staging.api.letsencrypt.org"
+        self.keys = monkey.gen_keys()
+        self.tempdir = tempfile.mkdtemp()
+        self.fuse_proc = Popen(["python", "tests/monkey.py", self.tempdir])
+
+    def tearDown(self):
+        self.fuse_proc.terminate()
+        self.fuse_proc.wait()
+        os.rmdir(self.tempdir)
 
-    def test_ca(self):
-        self.assertEqual(self.CA, "https://acme-staging.api.letsencrypt.org")
+    def test_success(self):
+        result = acme_tiny.get_crt(
+            self.keys['account_key'].name,
+            self.keys['domain_csr'].name,
+            self.tempdir,
+            CA=self.CA)