diff venv/lib/python2.7/site-packages/github/tests/Framework.py @ 0:d67268158946 draft

planemo upload commit a3f181f5f126803c654b3a66dd4e83a48f7e203b
author bcclaywell
date Mon, 12 Oct 2015 17:43:33 -0400
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/venv/lib/python2.7/site-packages/github/tests/Framework.py	Mon Oct 12 17:43:33 2015 -0400
@@ -0,0 +1,270 @@
+# -*- coding: utf-8 -*-
+
+# ########################## Copyrights and license ############################
+#                                                                              #
+# Copyright 2012 Vincent Jacques <vincent@vincent-jacques.net>                 #
+# Copyright 2012 Zearin <zearin@gonk.net>                                      #
+# Copyright 2013 AKFish <akfish@gmail.com>                                     #
+# Copyright 2013 Vincent Jacques <vincent@vincent-jacques.net>                 #
+#                                                                              #
+# This file is part of PyGithub. http://jacquev6.github.com/PyGithub/          #
+#                                                                              #
+# PyGithub is free software: you can redistribute it and/or modify it under    #
+# the terms of the GNU Lesser General Public License as published by the Free  #
+# Software Foundation, either version 3 of the License, or (at your option)    #
+# any later version.                                                           #
+#                                                                              #
+# PyGithub is distributed in the hope that it will be useful, but WITHOUT ANY  #
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS    #
+# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more #
+# details.                                                                     #
+#                                                                              #
+# You should have received a copy of the GNU Lesser General Public License     #
+# along with PyGithub. If not, see <http://www.gnu.org/licenses/>.             #
+#                                                                              #
+# ##############################################################################
+
+import os
+import sys
+import unittest
+import httplib
+import traceback
+
+import github
+
+atLeastPython26 = sys.hexversion >= 0x02060000
+atLeastPython3 = sys.hexversion >= 0x03000000
+atMostPython32 = sys.hexversion < 0x03030000
+
+if atLeastPython26:
+    import json
+else:  # pragma no cover (Covered by all tests with Python 2.5)
+    import simplejson as json  # pragma no cover (Covered by all tests with Python 2.5)
+
+
+def readLine(file):
+    if atLeastPython3:
+        return file.readline().decode("utf-8").strip()
+    else:
+        return file.readline().strip()
+
+
+class FakeHttpResponse:
+    def __init__(self, status, headers, output):
+        self.status = status
+        self.__headers = headers
+        self.__output = output
+
+    def getheaders(self):
+        return self.__headers
+
+    def read(self):
+        return self.__output
+
+
+def fixAuthorizationHeader(headers):
+    if "Authorization" in headers:
+        if headers["Authorization"].endswith("ZmFrZV9sb2dpbjpmYWtlX3Bhc3N3b3Jk"):
+            # This special case is here to test the real Authorization header
+            # sent by PyGithub. It would have avoided issue https://github.com/jacquev6/PyGithub/issues/153
+            # because we would have seen that Python 3 was not generating the same
+            # header as Python 2
+            pass
+        elif headers["Authorization"].startswith("token "):
+            headers["Authorization"] = "token private_token_removed"
+        elif headers["Authorization"].startswith("Basic "):
+            headers["Authorization"] = "Basic login_and_password_removed"
+
+
+class RecordingConnection:  # pragma no cover (Class useful only when recording new tests, not used during automated tests)
+    def __init__(self, file, protocol, host, port, *args, **kwds):
+        self.__file = file
+        self.__protocol = protocol
+        self.__host = host
+        self.__port = str(port)
+        self.__cnx = self._realConnection(host, port, *args, **kwds)
+
+    def request(self, verb, url, input, headers):
+        print verb, url, input, headers,
+        self.__cnx.request(verb, url, input, headers)
+        fixAuthorizationHeader(headers)
+        self.__writeLine(self.__protocol)
+        self.__writeLine(verb)
+        self.__writeLine(self.__host)
+        self.__writeLine(self.__port)
+        self.__writeLine(url)
+        self.__writeLine(str(headers))
+        self.__writeLine(input.replace('\n', '').replace('\r', ''))
+
+    def getresponse(self):
+        res = self.__cnx.getresponse()
+
+        status = res.status
+        print "=>", status
+        headers = res.getheaders()
+        output = res.read()
+
+        self.__writeLine(str(status))
+        self.__writeLine(str(headers))
+        self.__writeLine(str(output))
+
+        return FakeHttpResponse(status, headers, output)
+
+    def close(self):
+        self.__writeLine("")
+        return self.__cnx.close()
+
+    def __writeLine(self, line):
+        self.__file.write(line + "\n")
+
+
+class RecordingHttpConnection(RecordingConnection):  # pragma no cover (Class useful only when recording new tests, not used during automated tests)
+    _realConnection = httplib.HTTPConnection
+
+    def __init__(self, file, *args, **kwds):
+        RecordingConnection.__init__(self, file, "http", *args, **kwds)
+
+
+class RecordingHttpsConnection(RecordingConnection):  # pragma no cover (Class useful only when recording new tests, not used during automated tests)
+    _realConnection = httplib.HTTPSConnection
+
+    def __init__(self, file, *args, **kwds):
+        RecordingConnection.__init__(self, file, "https", *args, **kwds)
+
+
+class ReplayingConnection:
+    def __init__(self, testCase, file, protocol, host, port, *args, **kwds):
+        self.__testCase = testCase
+        self.__file = file
+        self.__protocol = protocol
+        self.__host = host
+        self.__port = str(port)
+
+    def request(self, verb, url, input, headers):
+        fixAuthorizationHeader(headers)
+        self.__testCase.assertEqual(self.__protocol, readLine(self.__file))
+        self.__testCase.assertEqual(verb, readLine(self.__file))
+        self.__testCase.assertEqual(self.__host, readLine(self.__file))
+        self.__testCase.assertEqual(self.__port, readLine(self.__file))
+        self.__testCase.assertEqual(self.__splitUrl(url), self.__splitUrl(readLine(self.__file)))
+        self.__testCase.assertEqual(headers, eval(readLine(self.__file)))
+        expectedInput = readLine(self.__file)
+        if input.startswith("{"):
+            self.__testCase.assertEqual(json.loads(input.replace('\n', '').replace('\r', '')), json.loads(expectedInput))
+        elif atMostPython32:  # @todo Test in all cases, including Python 3.3
+            # In Python 3.3, dicts are not output in the same order as in Python 2.5 -> 3.2.
+            # So, form-data encoding is not deterministic and is difficult to test.
+            self.__testCase.assertEqual(input.replace('\n', '').replace('\r', ''), expectedInput)
+
+    def __splitUrl(self, url):
+        splitedUrl = url.split("?")
+        if len(splitedUrl) == 1:
+            return splitedUrl
+        self.__testCase.assertEqual(len(splitedUrl), 2)
+        base, qs = splitedUrl
+        return (base, sorted(qs.split("&")))
+
+    def getresponse(self):
+        status = int(readLine(self.__file))
+        headers = eval(readLine(self.__file))
+        output = readLine(self.__file)
+
+        return FakeHttpResponse(status, headers, output)
+
+    def close(self):
+        readLine(self.__file)
+
+
+def ReplayingHttpConnection(testCase, file, *args, **kwds):
+    return ReplayingConnection(testCase, file, "http", *args, **kwds)
+
+
+def ReplayingHttpsConnection(testCase, file, *args, **kwds):
+    return ReplayingConnection(testCase, file, "https", *args, **kwds)
+
+
+class BasicTestCase(unittest.TestCase):
+    recordMode = False
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+        self.__fileName = ""
+        self.__file = None
+        if self.recordMode:  # pragma no cover (Branch useful only when recording new tests, not used during automated tests)
+            github.Requester.Requester.injectConnectionClasses(
+                lambda ignored, *args, **kwds: RecordingHttpConnection(self.__openFile("wb"), *args, **kwds),
+                lambda ignored, *args, **kwds: RecordingHttpsConnection(self.__openFile("wb"), *args, **kwds)
+            )
+            import GithubCredentials
+            self.login = GithubCredentials.login
+            self.password = GithubCredentials.password
+            self.oauth_token = GithubCredentials.oauth_token
+            # @todo Remove client_id and client_secret from ReplayData (as we already remove login, password and oauth_token)
+            # self.client_id = GithubCredentials.client_id
+            # self.client_secret = GithubCredentials.client_secret
+        else:
+            github.Requester.Requester.injectConnectionClasses(
+                lambda ignored, *args, **kwds: ReplayingHttpConnection(self, self.__openFile("rb"), *args, **kwds),
+                lambda ignored, *args, **kwds: ReplayingHttpsConnection(self, self.__openFile("rb"), *args, **kwds)
+            )
+            self.login = "login"
+            self.password = "password"
+            self.oauth_token = "oauth_token"
+            self.client_id = "client_id"
+            self.client_secret = "client_secret"
+
+    def tearDown(self):
+        unittest.TestCase.tearDown(self)
+        self.__closeReplayFileIfNeeded()
+        github.Requester.Requester.resetConnectionClasses()
+
+    def __openFile(self, mode):
+        for (_, _, functionName, _) in traceback.extract_stack():
+            if functionName.startswith("test") or functionName == "setUp" or functionName == "tearDown":
+                if functionName != "test":  # because in class Hook(Framework.TestCase), method testTest calls Hook.test
+                    fileName = os.path.join(os.path.dirname(__file__), "ReplayData", self.__class__.__name__ + "." + functionName + ".txt")
+        if fileName != self.__fileName:
+            self.__closeReplayFileIfNeeded()
+            self.__fileName = fileName
+            self.__file = open(self.__fileName, mode)
+        return self.__file
+
+    def __closeReplayFileIfNeeded(self):
+        if self.__file is not None:
+            if not self.recordMode:  # pragma no branch (Branch useful only when recording new tests, not used during automated tests)
+                self.assertEqual(readLine(self.__file), "")
+            self.__file.close()
+
+    def assertListKeyEqual(self, elements, key, expectedKeys):
+        realKeys = [key(element) for element in elements]
+        self.assertEqual(realKeys, expectedKeys)
+
+    def assertListKeyBegin(self, elements, key, expectedKeys):
+        realKeys = [key(element) for element in elements[: len(expectedKeys)]]
+        self.assertEqual(realKeys, expectedKeys)
+
+
+class TestCase(BasicTestCase):
+    def doCheckFrame(self, obj, frame):
+        if obj._headers == {} and frame is None:
+            return
+        if obj._headers is None and frame == {}:
+            return
+        self.assertEqual(obj._headers, frame[2])
+
+    def getFrameChecker(self):
+        return lambda requester, obj, frame: self.doCheckFrame(obj, frame)
+
+    def setUp(self):
+        BasicTestCase.setUp(self)
+
+        # Set up frame debugging
+        github.GithubObject.GithubObject.setCheckAfterInitFlag(True)
+        github.Requester.Requester.setDebugFlag(True)
+        github.Requester.Requester.setOnCheckMe(self.getFrameChecker())
+
+        self.g = github.Github(self.login, self.password)
+
+
+def activateRecordMode():  # pragma no cover (Function useful only when recording new tests, not used during automated tests)
+    BasicTestCase.recordMode = True