Permalink
Browse files

initial support for waiting for a device to reconnect with --stay-con…

…nected


fixes #6
  • Loading branch information...
1 parent 931790b commit 901b848af9cf211d3c5afd6e6446b648939df0ae @marshall committed Jul 2, 2013
Showing with 177 additions and 29 deletions.
  1. +50 −12 logcat-color
  2. +6 −0 logcatcolor/config.py
  3. +25 −4 test/common.py
  4. +15 −6 test/config_test.py
  5. +1 −0 test/configs/simple_config
  6. +30 −7 test/logcat_color_test.py
  7. +50 −0 test/mock-adb
View
@@ -14,10 +14,12 @@ import fcntl
import optparse
import os
import struct
-from subprocess import Popen, PIPE
import sys
import termios
+from colorama import Fore, Back, Style
+from subprocess import check_call, Popen, PIPE
+
from logcatcolor.config import LogcatColorConfig
from logcatcolor.profile import Profile
from logcatcolor.reader import LogcatReader
@@ -72,6 +74,10 @@ class LogcatColor(object):
parser.add_option("--no-wrap", action="store_false", dest="wrap",
default=None, help="don't wrap console text into a column " +
"(makes for better copy/paste)")
+ parser.add_option("--stay-connected", action="store_true", default=None,
+ dest="stay_connected", help="keep logcat-color running when the "
+ "device disconnects, and automatically "
+ "wait for the device to reconnect")
parser.add_option("-i", "--input", metavar="FILE", dest="input",
default=None,
help="read input from FILE, instead of starting adb. this is " +
@@ -187,28 +193,60 @@ class LogcatColor(object):
if self.format:
# put format in front in case custom filters are used
logcat_args[0:0] = ["-v", self.format]
- return logcat_args
- def start(self):
if self.profile:
buffers = self.profile.buffers
if buffers:
- for b in buffers: self.logcat_args.extend(["-b", b])
+ for b in buffers: logcat_args.extend(["-b", b])
- # if someone is piping, use stdin as input. if not, invoke adb logcat
- if self.input.isatty():
- adb_command = self.get_adb_args()
- adb_command.append("logcat")
- adb_command.extend(self.get_logcat_args())
- self.input = Popen(adb_command, stdout=PIPE).stdout
+ return logcat_args
+
+ def start_logcat(self):
+ adb_command = self.get_adb_args()
+ adb_command.append("logcat")
+ adb_command.extend(self.get_logcat_args())
+ self.input = Popen(adb_command, stdout=PIPE).stdout
+ def init_reader(self):
reader = LogcatReader(self.input, self.config, profile=self.profile,
format=self.format, layout=self.layout, writer=self.output,
width=self.width)
+
+ def start(self):
+ # if someone is piping, use stdin as input. if not, invoke adb logcat
+ if self.input.isatty():
+ self.start_logcat()
+
+ self.init_reader()
+
+ def loop(self):
try:
- asyncore.loop()
+ self.start()
+ while True:
+ asyncore.loop()
+ if not self.config.get_stay_connected():
+ break
+ self.wait_for_device()
+ self.start_logcat()
+ self.init_reader()
except KeyboardInterrupt, e:
pass
+ WAIT_FOR_DEVICE = Fore.WHITE + Back.BLACK + Style.DIM + \
+ "--- Waiting for device" + Style.RESET_ALL + \
+ Fore.BLUE + Back.BLACK + Style.DIM + " %s" + Style.RESET_ALL + \
+ Fore.WHITE + Back.BLACK + Style.DIM + "---" + Style.RESET_ALL
+
+ def wait_for_device(self):
+ command = self.get_adb_args()
+ command.append("wait-for-device")
+
+ device_str = ""
+ if self.adb_device:
+ device_str = "\"%s\" " % self.adb_device
+
+ print self.WAIT_FOR_DEVICE % device_str
+ check_call(command)
+
if __name__ == "__main__":
- LogcatColor().start()
+ LogcatColor().loop()
View
@@ -10,6 +10,7 @@ class LogcatColorConfig(object):
DEFAULT_LAYOUT = "brief"
DEFAULT_WRAP = True
DEFAULT_ADB = None
+ DEFAULT_STAY_CONNECTED = False
def __init__(self, options):
self.options = options
@@ -58,6 +59,8 @@ def get_default_config(self):
def post_load(self):
if self.options.wrap is not None:
self.config["wrap"] = self.options.wrap
+ if self.options.stay_connected is not None:
+ self.config["stay_connected"] = self.options.stay_connected
def get_default_layout(self):
return self.config.get("default_layout", self.DEFAULT_LAYOUT)
@@ -68,5 +71,8 @@ def get_column_width(self, column):
def get_wrap(self):
return self.config.get("wrap", self.DEFAULT_WRAP)
+ def get_stay_connected(self):
+ return self.config.get("stay_connected", self.DEFAULT_STAY_CONNECTED)
+
def get_adb(self):
return self.config.get("adb", self.DEFAULT_ADB)
View
@@ -2,16 +2,37 @@
import os
this_dir = os.path.abspath(os.path.dirname(__file__))
-test_dir = os.path.dirname(this_dir)
-filter_results = os.path.join(test_dir, ".filter_results")
+top_dir = os.path.dirname(this_dir)
+logcat_color = os.path.join(top_dir, "logcat-color")
+execfile(logcat_color)
+
+filter_results = os.path.join(this_dir, ".filter_results")
+mock_adb = os.path.join(this_dir, "mock-adb")
class MockObject(object):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
setattr(self, key, value)
-MockColumn = lambda name, def_width: MockObject(NAME=name, DEFAULT_WIDTH=def_width)
-MockOptions = lambda config, wrap: MockObject(config=config, wrap=wrap)
+class MockAdbLogcatColor(LogcatColor):
+ def __init__(self, log, results, args=None, max_wait_count=None):
+ LogcatColor.__init__(self, args=args)
+ self.log = log
+ self.results = results
+ self.wait_count = 0
+ self.max_wait_count = max_wait_count
+
+ def get_adb_args(self):
+ adb_args = LogcatColor.get_adb_args(self)
+ adb_args[0:1] = [mock_adb, "--log", self.log, "--results", self.results]
+ return adb_args
+
+ def wait_for_device(self):
+ LogcatColor.wait_for_device(self)
+ if self.max_wait_count:
+ self.wait_count += 1
+ if self.wait_count == self.max_wait_count:
+ raise KeyboardInterrupt()
def test_filter(fn):
def wrapped(data):
View
@@ -9,29 +9,38 @@
this_dir = os.path.abspath(os.path.dirname(__file__))
configs_dir = os.path.join(this_dir, "configs")
-def config_test(config_file, wrap=None):
+def config_test(config_file, wrap=None, stay_connected=None):
def run_config_test(fn):
def wrapped(self):
path = os.path.join(configs_dir, config_file)
- fn(self, LogcatColorConfig(MockOptions(path, wrap)))
+ options = MockObject(config=path,
+ wrap=wrap,
+ stay_connected=stay_connected)
+ fn(self, LogcatColorConfig(options))
return wrapped
return run_config_test
class ConfigTest(unittest.TestCase):
+ def setUp(self):
+ self.tag_column = MockObject(NAME="tag", DEFAULT_WIDTH=20)
+
@config_test("")
def test_default_config(self, config):
self.assertEqual(config.get_default_layout(), config.DEFAULT_LAYOUT)
- self.assertEqual(config.get_column_width(MockColumn("tag", 20)), 20)
+ self.assertEqual(config.get_column_width(self.tag_column), 20)
self.assertEqual(config.get_wrap(), config.DEFAULT_WRAP)
self.assertEqual(config.get_adb(), config.DEFAULT_ADB)
+ self.assertEqual(config.get_stay_connected(), config.DEFAULT_STAY_CONNECTED)
@config_test("simple_config")
def test_simple_config(self, config):
self.assertEqual(config.get_default_layout(), "test")
- self.assertEqual(config.get_column_width(MockColumn("tag", 20)), 1)
+ self.assertEqual(config.get_column_width(self.tag_column), 1)
self.assertFalse(config.get_wrap())
self.assertEqual(config.get_adb(), "/path/to/adb")
+ self.assertEqual(config.get_stay_connected(), True)
- @config_test("simple_config", wrap=True)
- def test_simple_config_override_wrap(self, config):
+ @config_test("simple_config", wrap=True, stay_connected=True)
+ def test_simple_config_overrides(self, config):
self.assertTrue(config.get_wrap())
+ self.assertTrue(config.get_stay_connected())
@@ -2,3 +2,4 @@ default_layout = "test"
tag_width = 1
wrap = False
adb = "/path/to/adb"
+stay_connected = True
View
@@ -1,14 +1,14 @@
import common
+import json
import os
from StringIO import StringIO
from subprocess import Popen, PIPE
import sys
import tempfile
import unittest
+from common import LogcatColor, MockAdbLogcatColor
this_dir = os.path.dirname(os.path.abspath(__file__))
-logcat_color = os.path.join(os.path.dirname(this_dir), "logcat-color")
-execfile(logcat_color)
def logcat_color_test(*args, **kwargs):
def run_logcat_color_test(fn):
@@ -31,14 +31,14 @@ def wrapped(self):
class LogcatColorTest(unittest.TestCase):
DEBUG = False
- @classmethod
- def tearDownClass(cls):
+ def setUp(self):
+ # Clear out our temporary output file before each test
global tmpout
- os.unlink(tmpout)
+ with open(tmpout, "w") as f: f.write("")
def start_logcat_color(self, *args, **kwargs):
args = list(args)
- args.insert(0, logcat_color)
+ args.insert(0, common.logcat_color)
if "config" in kwargs:
args[1:1] = ["--config", kwargs["config"]]
del kwargs["config"]
@@ -140,7 +140,7 @@ def test_logcat_options_with_filters(self):
# Make sure logcat flags come before filter arguments
# https://github.com/marshall/logcat-color/issues/5
lc = LogcatColor(args=["-v", "time", "Tag1:V", "*:S", "--silent",
- "--print-size", "--dump", "--clear"])
+ "--print-size", "--dump", "--clear"])
self.assertEqual(lc.format, "time")
args = lc.get_logcat_args()
@@ -157,3 +157,26 @@ def test_logcat_options_with_filters(self):
self.assertEqual(args[-2], "Tag1:V")
self.assertEqual(args[-1], "*:S")
+
+ def test_stay_connected(self):
+ lc = MockAdbLogcatColor(BRIEF_LOG, tmpout,
+ args=["-s", "serial123", "--stay-connected",
+ "--config", EMPTY_CONFIG],
+ max_wait_count=3)
+ self.assertEqual(lc.config.get_stay_connected(), True)
+
+ lc.loop()
+ self.assertEqual(lc.wait_count, 3)
+
+ results = json.loads(open(tmpout, "r").read())
+ self.assertEqual(len(results), 6)
+
+ logcat_results = filter(lambda d: d["command"] == "logcat", results)
+ self.assertEqual(len(logcat_results), 3)
+
+ wait_results = filter(lambda d: d["command"] == "wait-for-device", results)
+ self.assertEquals(len(wait_results), 3)
+
+ for r in results:
+ self.assertEqual(r["serial"], "serial123")
+
View
@@ -0,0 +1,50 @@
+#!/usr/bin/env python
+import argparse
+import json
+import os
+import sys
+import time
+
+class MockAdb(object):
+ def __init__(self, args, command_args):
+ self.results_data = {}
+ for attr, value in args.__dict__.iteritems():
+ setattr(self, attr, value)
+ self.results_data[attr] = value
+
+ self.results_data["command_args"] = self.command_args = command_args
+
+ def run(self):
+ cmd = self.command.replace("-", "_")
+ if hasattr(self, cmd):
+ getattr(self, cmd)()
+
+ data = []
+ if os.path.exists(self.results):
+ try:
+ data = json.loads(open(self.results, "r").read())
+ except: pass
+
+ data.append(self.results_data)
+
+ with open(self.results, "w") as f:
+ f.write(json.dumps(data))
+
+ def wait_for_device(self):
+ time.sleep(0.1)
+
+ def logcat(self):
+ print open(self.log, "r").read()
+
+def main():
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--log")
+ parser.add_argument("--results")
+ parser.add_argument("command")
+ parser.add_argument("-d", dest="device", action="store_true", default=False)
+ parser.add_argument("-e", dest="emulator", action="store_true", default=False)
+ parser.add_argument("-s", dest="serial", default=None)
+ MockAdb(*parser.parse_known_args()).run()
+
+if __name__ == "__main__":
+ main()

0 comments on commit 901b848

Please sign in to comment.