#!/usr/bin/env python3

import pipes
import re
import subprocess
import unittest

BITNESS_32 = ("", "32")
BITNESS_64 = ("64", "64")

NATIVE_TEST_CLIENT_FOR_BITNESS = ' /data/nativetest%s/aidl_test_client/aidl_test_client%s'
NATIVE_TEST_SERVICE_FOR_BITNESS = ' /data/nativetest%s/aidl_test_service/aidl_test_service%s'

# From tools/base/ddmlib/src/main/java/com/android/ddmlib/testrunner/InstrumentationResultParser.java
INSTRUMENTATION_FAILURES_PATTERN = r'There (was|were) \d+ failure'

class TestFail(Exception):
    """Raised on test failures."""
    pass

def pretty_bitness(bitness):
    """Returns a human readable version of bitness, corresponding to BITNESS_* variable"""
    return bitness[-1]

class ShellResult(object):
    """Represents the result of running a shell command."""

    def __init__(self, exit_status, stdout, stderr):
        """Construct an instance.

        Args:
            exit_status: integer exit code of shell command
            stdout: string stdout of shell command
            stderr: string stderr of shell command
        """
        self.stdout = stdout
        self.stderr = stderr
        self.exit_status = exit_status

    def printable_string(self):
        """Get a string we could print to the logs and understand."""
        output = []
        output.append('stdout:')
        for line in self.stdout.splitlines():
            output.append('  > %s' % line)
        output.append('stderr:')
        for line in self.stderr.splitlines():
            output.append('  > %s' % line)
        return '\n'.join(output)


class AdbHost(object):
    """Represents a device connected via ADB."""

    def run(self, command, background=False, ignore_status=False):
        """Run a command on the device via adb shell.

        Args:
            command: string containing a shell command to run.
            background: True iff we should run this command in the background.
            ignore_status: True iff we should ignore the command's exit code.

        Returns:
            instance of ShellResult.

        Raises:
            subprocess.CalledProcessError on command exit != 0.
        """
        if background:
            command = '( %s ) </dev/null >/dev/null 2>&1 &' % command
        return self.adb('shell %s' % pipes.quote(command),
                        ignore_status=ignore_status)

    def adb(self, command, ignore_status=False):
        """Run an ADB command (e.g. `adb sync`).

        Args:
            command: string containing command to run
            ignore_status: True iff we should ignore the command's exit code.

        Returns:
            instance of ShellResult.

        Raises:
            subprocess.CalledProcessError on command exit != 0.
        """
        command = 'adb %s' % command
        p = subprocess.Popen(command, shell=True, close_fds=True,
                             stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                             universal_newlines=True)
        stdout, stderr = p.communicate()
        if not ignore_status and p.returncode:
            raise subprocess.CalledProcessError(p.returncode, command)
        return ShellResult(p.returncode, stdout, stderr)

class NativeServer:
    def __init__(self, host, bitness):
        self.name = "%s_bit_native_server" % pretty_bitness(bitness)
        self.host = host
        self.binary = NATIVE_TEST_SERVICE_FOR_BITNESS % bitness
    def cleanup(self):
        self.host.run('killall %s' % self.binary, ignore_status=True)
    def run(self):
        return self.host.run(self.binary, background=True)

class NativeClient:
    def __init__(self, host, bitness):
        self.name = "%s_bit_native_client" % pretty_bitness(bitness)
        self.host = host
        self.binary = NATIVE_TEST_CLIENT_FOR_BITNESS % bitness
    def cleanup(self):
        self.host.run('killall %s' % self.binary, ignore_status=True)
    def run(self):
        result = self.host.run(self.binary + ' --gtest_color=yes', ignore_status=True)
        print(result.printable_string())
        if result.exit_status:
            raise TestFail(result.stdout)

class JavaClient:
    def __init__(self, host, native_bitness):
        self.name = "java_client"
        self.host = host
        self.native_bitness = native_bitness
    def cleanup(self):
        host.run('setenforce 1')
        self.host.run('killall android.aidl.tests', ignore_status=True)
    def run(self):
        host.run('setenforce 0') # Java app needs selinux off
        result = self.host.run('am instrument -w --no-hidden-api-checks '
                               'android.aidl.tests/'
                               'androidx.test.runner.AndroidJUnitRunner')
        print(result.printable_string())
        if re.search(INSTRUMENTATION_FAILURES_PATTERN, result.stdout) is not None:
            raise TestFail(result.stdout)

def supported_bitnesses(host):
    bitnesses = []
    for bitness in [BITNESS_32, BITNESS_64]:
        native_client = NATIVE_TEST_CLIENT_FOR_BITNESS % bitness
        if host.run('ls %s' % native_client, ignore_status=True).exit_status == 0:
            bitnesses += [bitness]
    return bitnesses

# tests added dynamically below
class TestAidl(unittest.TestCase):
    pass

def make_test(client, server):
    def test(self):
        try:
            client.cleanup()
            server.cleanup()
            server.run()
            client.run()
        finally:
            client.cleanup()
            server.cleanup()
    return test

if __name__ == '__main__':
    host = AdbHost()
    bitnesses = supported_bitnesses(host)
    if len(bitnesses) == 0:
        print("No clients installed")
        exit(1)

    clients = []
    servers = []

    for bitness in bitnesses:
        clients += [NativeClient(host, bitness)]
        servers += [NativeServer(host, bitness)]

    # Java only supports one bitness, but needs to run a native binary
    # to process its results
    clients += [JavaClient(host, bitnesses[-1])]

    for client in clients:
        for server in servers:
            test_name = 'test_%s_to_%s' % (client.name, server.name)
            test = make_test(client, server)
            setattr(TestAidl, test_name, test)

    suite = unittest.TestLoader().loadTestsFromTestCase(TestAidl)
    unittest.TextTestRunner(verbosity=2).run(suite)