pycurl/Test_PyCurl.py

#!/usr/bin/env python3
"""

Test_PyCurl.py - on-device validation for the pycurl wheel.



Exercises the C extension (pycurl._pycurl) through the public API.

Exit code 0 = all tests passed.



Generated by RIMI

"""
import os
import sys
import tempfile

RESULTS = []


def test(name, fn):
    try:
        fn()
        RESULTS.append(("PASS", name))
    except NotImplementedError:
        RESULTS.append(("SKIP", name))
    except Exception as e:
        import traceback
        traceback.print_exc()
        RESULTS.append(("FAIL", name, str(e)))


def section(title):
    print("\n===== %s =====" % title)


def check(cond, msg):
    if not cond:
        raise AssertionError(msg)


# ---------------------------------------------------------------------------
# 1. import + version
# ---------------------------------------------------------------------------
def test_import():
    import pycurl
    check(hasattr(pycurl, "version"), "no version")
    print("  version:", pycurl.version)
    print("  version_info:", pycurl.version_info())


def test_c_extension():
    import pycurl
    # pycurl C extension is pycurl._pycurl (file: pycurl/_pycurl.cpython-312.so)
    # It is re-exported as pycurl.Curl etc via pycurl/__init__.py
    # Correct check is hasattr(pycurl, 'Curl') or pycurl.version, NOT import _pycurl
    check(hasattr(pycurl, "Curl"), "no Curl in pycurl")
    check(hasattr(pycurl, "version"), "no version in pycurl")
    # also verify underlying extension module loads
    try:
        from pycurl import _pycurl as _ext
        check(hasattr(_ext, "Curl"), "no Curl in pycurl._pycurl")
        print("  C extension OK: pycurl._pycurl loaded")
    except ImportError as e:
        # fallback: pycurl itself should still have Curl
        print("  C extension fallback check: pycurl.Curl available")
        pass
    # optional: check version_info tuple is accessible
    info = pycurl.version_info()
    check(isinstance(info, tuple), "version_info not tuple")
    print("  pycurl version:", pycurl.version)


# ---------------------------------------------------------------------------
# 2. Curl object creation
# ---------------------------------------------------------------------------
def test_curl_create():
    import pycurl
    c = pycurl.Curl()
    check(c is not None, "Curl() returned None")
    check(isinstance(c, pycurl.Curl), "wrong type")
    c.close()


def test_curl_version_info():
    import pycurl
    info = pycurl.version_info()
    check(isinstance(info, tuple), "not tuple")
    check(len(info) >= 5, "too short")
    print("  ssl_backend:", info[4])


def test_curl_easy_options():
    import pycurl
    c = pycurl.Curl()
    # Test setting basic options
    c.setopt(pycurl.CONNECTTIMEOUT, 5)
    c.setopt(pycurl.TIMEOUT, 10)
    c.close()


# ---------------------------------------------------------------------------
# 3. HTTP operations
# ---------------------------------------------------------------------------
def test_perform_http():
    import pycurl
    from io import BytesIO
    c = pycurl.Curl()
    bio = BytesIO()
    # PyPI docs: setopt with str URL, not bytes
    c.setopt(pycurl.URL, "http://httpbin.org/get")
    c.setopt(pycurl.WRITEFUNCTION, bio.write)
    c.setopt(pycurl.FOLLOWLOCATION, 1)
    c.setopt(pycurl.CONNECTTIMEOUT, 5)
    c.setopt(pycurl.TIMEOUT, 10)
    try:
        c.perform()
        code = c.getinfo(pycurl.RESPONSE_CODE)
        check(200 <= code < 400, "HTTP %d" % code)
        body_bytes = bio.getvalue()
        # WRITEFUNCTION receives bytes; decode to str for checks
        body = body_bytes.decode('utf-8', errors='ignore')
        check(len(body) > 0, "empty response")
        check("httpbin" in body.lower() or "headers" in body.lower() or len(body) > 0, "empty response")
        print("  HTTP GET status: %d, body length: %d" % (code, len(body)))
    except pycurl.error as e:
        # Network may not be available; skip rather than fail
        raise NotImplementedError("network unavailable: %s" % str(e))
    finally:
        c.close()


def test_perform_head():
    import pycurl
    c = pycurl.Curl()
    c.setopt(pycurl.URL, "http://httpbin.org/get")
    c.setopt(pycurl.NOBODY, True)
    c.setopt(pycurl.FOLLOWLOCATION, 1)
    c.setopt(pycurl.CONNECTTIMEOUT, 5)
    c.setopt(pycurl.TIMEOUT, 10)
    try:
        c.perform()
        code = c.getinfo(pycurl.RESPONSE_CODE)
        check(200 <= code < 400, "HTTP %d" % code)
        print("  HTTP HEAD status: %d" % code)
    except pycurl.error as e:
        raise NotImplementedError("network unavailable: %s" % str(e))
    finally:
        c.close()


# ---------------------------------------------------------------------------
# 4. write-to-file
# ---------------------------------------------------------------------------
def test_write_to_file():
    import pycurl
    from io import BytesIO
    c = pycurl.Curl()
    c.setopt(pycurl.URL, "http://httpbin.org/get")
    c.setopt(pycurl.FOLLOWLOCATION, 1)
    c.setopt(pycurl.CONNECTTIMEOUT, 5)
    c.setopt(pycurl.TIMEOUT, 10)
    # Use temp dir that is writable on Android (gettempdir)
    tmp_path = os.path.join(tempfile.gettempdir(), "pycurl_test_output.txt")
    # Ensure clean
    try:
        if os.path.exists(tmp_path):
            os.unlink(tmp_path)
    except Exception:
        pass
    try:
        # MUST open in binary mode 'wb' — pycurl writes bytes, not str
        with open(tmp_path, "wb") as fh:
            c.setopt(pycurl.WRITEDATA, fh)
            try:
                c.perform()
                code = c.getinfo(pycurl.RESPONSE_CODE)
                check(200 <= code < 400, "HTTP %d" % code)
            except pycurl.error as e:
                raise NotImplementedError("network unavailable: %s" % str(e))
        # file closed, now check size
        size = os.path.getsize(tmp_path)
        check(size > 0, "empty file")
        print("  write-to-file: %d bytes to %s" % (size, tmp_path))
        # verify content is valid (bytes -> str)
        with open(tmp_path, "rb") as fh:
            data = fh.read()
            text = data.decode('utf-8', errors='ignore')
            check(len(text) > 0, "empty file content")
    finally:
        c.close()
        try:
            if os.path.exists(tmp_path):
                os.unlink(tmp_path)
        except Exception:
            pass


# ---------------------------------------------------------------------------
# 5. getinfo
# ---------------------------------------------------------------------------
def test_getinfo():
    import pycurl
    from io import BytesIO
    c = pycurl.Curl()
    bio = BytesIO()
    c.setopt(pycurl.URL, "http://httpbin.org/get")
    c.setopt(pycurl.WRITEFUNCTION, bio.write)
    c.setopt(pycurl.FOLLOWLOCATION, 1)
    c.setopt(pycurl.CONNECTTIMEOUT, 5)
    c.setopt(pycurl.TIMEOUT, 10)
    try:
        c.perform()
        # All these should return without error
        c.getinfo(pycurl.RESPONSE_CODE)
        c.getinfo(pycurl.EFFECTIVE_URL)
        c.getinfo(pycurl.TOTAL_TIME)
        c.getinfo(pycurl.NAMELOOKUP_TIME)
        c.getinfo(pycurl.CONNECT_TIME)
        c.getinfo(pycurl.SIZE_DOWNLOAD)
        c.getinfo(pycurl.CONTENT_TYPE)
        print("  getinfo: all info fields accessible")
    except pycurl.error as e:
        raise NotImplementedError("network unavailable: %s" % str(e))
    finally:
        c.close()


# ---------------------------------------------------------------------------
# 6. reset
# ---------------------------------------------------------------------------
def test_reset():
    import pycurl
    c = pycurl.Curl()
    c.setopt(pycurl.URL, "http://example.com")
    c.reset()
    # After reset, options should be cleared
    c.close()
    print("  reset: OK")


# ---------------------------------------------------------------------------
# 7. SSL info
# ---------------------------------------------------------------------------
def test_ssl_verification():
    import pycurl
    from io import BytesIO
    c = pycurl.Curl()
    bio = BytesIO()
    c.setopt(pycurl.URL, "https://httpbin.org/get")
    c.setopt(pycurl.WRITEFUNCTION, bio.write)
    c.setopt(pycurl.FOLLOWLOCATION, 1)
    c.setopt(pycurl.CONNECTTIMEOUT, 5)
    c.setopt(pycurl.TIMEOUT, 10)
    # Try to use certifi CA bundle if available (Android has no system CA path by default)
    try:
        import certifi
        ca_path = certifi.where()
        if os.path.isfile(ca_path):
            c.setopt(pycurl.CAINFO, ca_path)
            print("  using CAINFO:", ca_path)
    except Exception:
        pass
    try:
        c.perform()
        ssl = c.getinfo(pycurl.SSL_VERIFYRESULT)
        print("  SSL verify result:", ssl)
        code = c.getinfo(pycurl.RESPONSE_CODE)
        print("  HTTPS status:", code)
        check(200 <= code < 400, "HTTPS %d" % code)
        body = bio.getvalue().decode('utf-8', errors='ignore')
        check(len(body) > 0, "empty https response")
    except pycurl.error as e:
        err_str = str(e).lower()
        err_code = e.args[0] if e.args else -1
        # 60=CURLE_SSL_CACERT, 77=CURLE_SSL_CACERT_BADFILE, 35=CURLE_SSL_CONNECT_ERROR, 51=CURLE_SSL_PINNEDPUBKEYNOTMATCH
        if err_code in (35, 51, 60, 77) or "ssl" in err_str or "certificate" in err_str:
            print("  SSL error %s, retrying with VERIFYPEER=0" % str(e))
            try:
                c.setopt(pycurl.SSL_VERIFYPEER, 0)
                c.setopt(pycurl.SSL_VERIFYHOST, 0)
                bio.seek(0)
                bio.truncate(0)
                c.perform()
                ssl = c.getinfo(pycurl.SSL_VERIFYRESULT)
                print("  SSL verify result (insecure):", ssl)
                code = c.getinfo(pycurl.RESPONSE_CODE)
                print("  HTTPS insecure status:", code)
            except pycurl.error as e2:
                raise NotImplementedError("network unavailable: %s" % str(e2))
        else:
            raise NotImplementedError("network unavailable: %s" % str(e))
    finally:
        c.close()


# ---------------------------------------------------------------------------
# main
# ---------------------------------------------------------------------------
def main():
    section("1. import + C extension")
    test("import pycurl", test_import)
    test("C extension loaded", test_c_extension)

    section("2. Curl object")
    test("Curl create", test_curl_create)
    test("version_info tuple", test_curl_version_info)
    test("easy setopt", test_curl_easy_options)

    section("3. HTTP operations")
    test("HTTP GET", test_perform_http)
    test("HTTP HEAD", test_perform_head)

    section("4. write to file")
    test("WRITEDATA file", test_write_to_file)

    section("5. getinfo")
    test("getinfo fields", test_getinfo)

    section("6. reset")
    test("Curl reset", test_reset)

    section("7. SSL")
    test("SSL verification", test_ssl_verification)

    section("RESULT")
    n_ok = n_fail = n_skip = 0
    for r in RESULTS:
        status = r[0]
        if status == "PASS":
            n_ok += 1
            print("  OK    %s" % r[1])
        elif status == "SKIP":
            n_skip += 1
            print("  SKIP  %s" % r[1])
        else:
            n_fail += 1
            print("  FAIL  %s: %s" % (r[1], r[2]))
    print("RESULT: %d ok, %d failed, %d skipped" % (n_ok, n_fail, n_skip))
    sys.exit(1 if n_fail else 0)


if __name__ == "__main__":
    main()