Use io.open to set file encoding (#465)
diff --git a/mobly/config_parser.py b/mobly/config_parser.py
index a61bfe1..4eed6a1 100644
--- a/mobly/config_parser.py
+++ b/mobly/config_parser.py
@@ -15,6 +15,7 @@
from builtins import str
import copy
+import io
import os
import yaml
@@ -149,7 +150,7 @@
Returns:
A dict that represents info in the config file.
"""
- with open(utils.abs_path(path), 'r') as f:
+ with io.open(utils.abs_path(path), 'r', encoding='utf-8') as f:
conf = yaml.load(f)
return conf
diff --git a/mobly/controllers/android_device.py b/mobly/controllers/android_device.py
index 2ff854e..504f671 100644
--- a/mobly/controllers/android_device.py
+++ b/mobly/controllers/android_device.py
@@ -13,10 +13,10 @@
# limitations under the License.
from builtins import str as new_str
-from builtins import open
from past.builtins import basestring
import contextlib
+import io
import logging
import os
import shutil
@@ -436,9 +436,8 @@
self._log_path = os.path.join(self._log_path_base,
'AndroidDevice%s' % self._serial)
self._debug_tag = self._serial
- self.log = AndroidDeviceLoggerAdapter(logging.getLogger(), {
- 'tag': self.debug_tag
- })
+ self.log = AndroidDeviceLoggerAdapter(logging.getLogger(),
+ {'tag': self.debug_tag})
self.sl4a = None
self.ed = None
self._adb_logcat_process = None
@@ -964,9 +963,10 @@
tag = tag[:tag_len]
out_name = tag + out_name
full_adblog_path = os.path.join(adb_excerpt_path, out_name)
- with open(full_adblog_path, 'w', encoding='utf-8') as out:
+ with io.open(full_adblog_path, 'w', encoding='utf-8') as out:
in_file = self.adb_logcat_file_path
- with open(in_file, 'r', encoding='utf-8', errors='replace') as f:
+ with io.open(
+ in_file, 'r', encoding='utf-8', errors='replace') as f:
in_range = False
while True:
line = None
diff --git a/mobly/controllers/iperf_server.py b/mobly/controllers/iperf_server.py
index c72d70e..4d01e17 100644
--- a/mobly/controllers/iperf_server.py
+++ b/mobly/controllers/iperf_server.py
@@ -1,17 +1,18 @@
# Copyright 2016 Google Inc.
-#
+#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
-#
+#
# http://www.apache.org/licenses/LICENSE-2.0
-#
+#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
+import io
import json
import logging
import os
@@ -43,10 +44,10 @@
class IPerfResult(object):
def __init__(self, result_path):
try:
- with open(result_path, 'r') as f:
+ with io.open(result_path, 'r', encoding='utf-8') as f:
self.result = json.load(f)
except ValueError:
- with open(result_path, 'r') as f:
+ with io.open(result_path, 'r', encoding='utf-8') as f:
# Possibly a result from interrupted iperf run, skip first line
# and try again.
lines = f.readlines()[1:]
@@ -135,8 +136,8 @@
utils.create_dir(self.log_path)
if tag:
tag = tag + ','
- out_file_name = "IPerfServer,{},{}{}.log".format(self.port, tag,
- len(self.log_files))
+ out_file_name = "IPerfServer,{},{}{}.log".format(
+ self.port, tag, len(self.log_files))
full_out_path = os.path.join(self.log_path, out_file_name)
cmd = '%s %s > %s' % (self.iperf_str, extra_args, full_out_path)
self.iperf_process = utils.start_standing_subprocess(cmd, shell=True)
diff --git a/mobly/controllers/monsoon.py b/mobly/controllers/monsoon.py
index 60fc69f..314bfbd 100644
--- a/mobly/controllers/monsoon.py
+++ b/mobly/controllers/monsoon.py
@@ -11,7 +11,6 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-
"""Interface for a USB-connected Monsoon power meter
(http://msoon.com/LabEquipment/PowerMonitor/).
"""
@@ -19,6 +18,7 @@
from past.builtins import basestring
import fcntl
+import io
import logging
import os
import select
@@ -143,7 +143,7 @@
if not dev.startswith(prefix):
continue
tmpname = "/tmp/monsoon.%s.%s" % (os.uname()[0], dev)
- self._tempfile = open(tmpname, "w")
+ self._tempfile = open(tmpname, "w", encoding='utf-8')
try:
os.chmod(tmpname, 0o666)
except OSError as e:
@@ -240,8 +240,8 @@
logging.warning("Wanted status, dropped type=0x%02x, len=%d",
read_bytes[0], len(read_bytes))
continue
- status = dict(zip(STATUS_FIELDS, struct.unpack(STATUS_FORMAT,
- read_bytes)))
+ status = dict(
+ zip(STATUS_FIELDS, struct.unpack(STATUS_FORMAT, read_bytes)))
p_type = status["packetType"]
if p_type != 0x10:
raise MonsoonError("Package type %s is not 0x10." % p_type)
@@ -344,8 +344,11 @@
continue
seq, _type, x, y = struct.unpack("BBBB", _bytes[:4])
- data = [struct.unpack(">hhhh", _bytes[x:x + 8])
- for x in range(4, len(_bytes) - 8, 8)]
+ data = [
+ struct.unpack(">hhhh", _bytes[x:x + 8])
+ for x in range(4,
+ len(_bytes) - 8, 8)
+ ]
if self._last_seq and seq & 0xF != (self._last_seq + 1) & 0xF:
logging.warning("Data sequence skipped, lost packet?")
@@ -421,8 +424,8 @@
self.ser.flush()
flushed = 0
while True:
- ready_r, ready_w, ready_x = select.select(
- [self.ser], [], [self.ser], 0)
+ ready_r, ready_w, ready_x = select.select([self.ser], [],
+ [self.ser], 0)
if len(ready_x) > 0:
logging.error("Exception from serial port.")
return None
@@ -464,9 +467,9 @@
self.offset = offset
num_of_data_pt = len(self._data_points)
if self.offset >= num_of_data_pt:
- raise MonsoonError(("Offset number (%d) must be smaller than the "
- "number of data points (%d).") %
- (offset, num_of_data_pt))
+ raise MonsoonError(
+ ("Offset number (%d) must be smaller than the "
+ "number of data points (%d).") % (offset, num_of_data_pt))
self.data_points = self._data_points[self.offset:]
self.timestamps = self._timestamps[self.offset:]
self.hz = hz
@@ -512,11 +515,12 @@
lines = data_str.strip().split('\n')
err_msg = ("Invalid input string format. Is this string generated by "
"MonsoonData class?")
- conditions = [len(lines) <= 4, "Average Current:" not in lines[1],
- "Voltage: " not in lines[2],
- "Total Power: " not in lines[3],
- "samples taken at " not in lines[4],
- lines[5] != "Time" + ' ' * 7 + "Amp"]
+ conditions = [
+ len(lines) <= 4, "Average Current:" not in lines[1],
+ "Voltage: " not in lines[2], "Total Power: " not in lines[3],
+ "samples taken at " not in lines[4],
+ lines[5] != "Time" + ' ' * 7 + "Amp"
+ ]
if any(conditions):
raise MonsoonError(err_msg)
hz_str = lines[4].split()[2]
@@ -549,7 +553,7 @@
raise MonsoonError("Attempting to write empty Monsoon data to "
"file, abort")
utils.create_dir(os.path.dirname(file_path))
- with open(file_path, 'w') as f:
+ with io.open(file_path, 'w', encoding='utf-8') as f:
for md in monsoon_data:
f.write(str(md))
f.write(MonsoonData.delimiter)
@@ -567,7 +571,7 @@
A list of MonsoonData objects.
"""
results = []
- with open(file_path, 'r') as f:
+ with io.open(file_path, 'r', encoding='utf-8') as f:
data_strs = f.read().split(MonsoonData.delimiter)
for data_str in data_strs:
results.append(MonsoonData.from_string(data_str))
@@ -634,9 +638,9 @@
strs.append("Average Current: {}mA.".format(self.average_current))
strs.append("Voltage: {}V.".format(self.voltage))
strs.append("Total Power: {}mW.".format(self.total_power))
- strs.append(("{} samples taken at {}Hz, with an offset of {} samples."
- ).format(
- len(self._data_points), self.hz, self.offset))
+ strs.append(
+ ("{} samples taken at {}Hz, with an offset of {} samples.").format(
+ len(self._data_points), self.hz, self.offset))
return "\n".join(strs)
def __len__(self):
@@ -794,11 +798,12 @@
pass
self.mon.StopDataCollection()
try:
- return MonsoonData(current_values,
- timestamps,
- sample_hz,
- voltage,
- offset=sample_offset)
+ return MonsoonData(
+ current_values,
+ timestamps,
+ sample_hz,
+ voltage,
+ offset=sample_offset)
except:
return None
@@ -864,8 +869,8 @@
try:
data = self.take_samples(hz, num, sample_offset=oset)
if not data:
- raise MonsoonError("No data was collected in measurement %s." %
- tag)
+ raise MonsoonError(
+ "No data was collected in measurement %s." % tag)
data.tag = tag
self.dut.log.info("Measurement summary: %s", repr(data))
return data
diff --git a/mobly/records.py b/mobly/records.py
index 74dd8f9..ad37cad 100644
--- a/mobly/records.py
+++ b/mobly/records.py
@@ -14,11 +14,10 @@
"""This module has classes for test result collection, and test result output.
"""
-from io import open
-
import collections
import copy
import enum
+import io
import logging
import sys
import threading
@@ -121,7 +120,7 @@
# because Python3 file descriptors set an encoding by default, which
# PyYAML uses instead of the encoding on yaml.safe_dump. So, the
# encoding has to be set on the open call instead.
- with open(self._path, 'a', encoding='utf-8') as f:
+ with io.open(self._path, 'a', encoding='utf-8') as f:
# Use safe_dump here to avoid language-specific tags in final
# output.
yaml.safe_dump(
diff --git a/mobly/utils.py b/mobly/utils.py
index 275d990..b4f8536 100644
--- a/mobly/utils.py
+++ b/mobly/utils.py
@@ -15,6 +15,7 @@
import base64
import concurrent.futures
import datetime
+import io
import logging
import os
import platform
@@ -209,7 +210,7 @@
A base64 string representing the content of the file in utf-8 encoding.
"""
path = abs_path(f_path)
- with open(path, 'rb') as f:
+ with io.open(path, 'rb') as f:
f_bytes = f.read()
base64_str = base64.b64encode(f_bytes).decode("utf-8")
return base64_str
diff --git a/tests/mobly/base_test_test.py b/tests/mobly/base_test_test.py
index a20e1fb..593ff0b 100755
--- a/tests/mobly/base_test_test.py
+++ b/tests/mobly/base_test_test.py
@@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+import io
import os
import mock
import shutil
@@ -1734,7 +1735,7 @@
actual_record = bt_cls.results.passed[0]
self.assertEqual(actual_record.test_name, "test_something")
hit = False
- with open(self.summary_file, 'r') as f:
+ with io.open(self.summary_file, 'r', encoding='utf-8') as f:
for c in yaml.load_all(f):
if c['Type'] != records.TestSummaryEntryType.USER_DATA.value:
continue
diff --git a/tests/mobly/config_parser_test.py b/tests/mobly/config_parser_test.py
new file mode 100644
index 0000000..ea16668
--- /dev/null
+++ b/tests/mobly/config_parser_test.py
@@ -0,0 +1,61 @@
+# Copyright 2018 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import io
+import os
+import shutil
+import tempfile
+import unittest
+
+from mobly import config_parser
+
+
+class OutputTest(unittest.TestCase):
+ """This test class has unit tests for the implementation of Mobly's output
+ files.
+ """
+
+ def setUp(self):
+ self.tmp_dir = tempfile.mkdtemp()
+
+ def tearDown(self):
+ shutil.rmtree(self.tmp_dir)
+
+ def test__load_config_file(self):
+ tmp_file_path = os.path.join(self.tmp_dir, 'config.yml')
+ with io.open(tmp_file_path, 'w', encoding='utf-8') as f:
+ f.write(u'TestBeds:\n')
+ f.write(u' # A test bed where adb will find Android devices.\n')
+ f.write(u' - Name: SampleTestBed\n')
+ f.write(u' Controllers:\n')
+ f.write(u' AndroidDevice: \'*\'\n')
+
+ config = config_parser._load_config_file(tmp_file_path)
+ self.assertEqual(config['TestBeds'][0]['Name'], u'SampleTestBed')
+
+ def test__load_config_file_with_unicode(self):
+ tmp_file_path = os.path.join(self.tmp_dir, 'config.yml')
+ with io.open(tmp_file_path, 'w', encoding='utf-8') as f:
+ f.write(u'TestBeds:\n')
+ f.write(u' # A test bed where adb will find Android devices.\n')
+ f.write(u' - Name: \u901a\n')
+ f.write(u' Controllers:\n')
+ f.write(u' AndroidDevice: \'*\'\n')
+
+ config = config_parser._load_config_file(tmp_file_path)
+ self.assertEqual(config['TestBeds'][0]['Name'], u'\u901a')
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_test.py b/tests/mobly/controllers/android_device_test.py
index 6949903..e12e511 100755
--- a/tests/mobly/controllers/android_device_test.py
+++ b/tests/mobly/controllers/android_device_test.py
@@ -14,6 +14,7 @@
from builtins import str as new_str
+import io
import logging
import mock
import os
@@ -38,11 +39,23 @@
'02-29 14:02:21.456 4454 Something\n',
'02-29 14:02:21.789 4454 Something again\n'
]
-# A mockd piece of adb logcat output.
-MOCK_ADB_LOGCAT = ('02-29 14:02:19.123 4454 Nothing\n'
- '%s'
- '02-29 14:02:22.123 4454 Something again and again\n'
- ) % ''.join(MOCK_ADB_LOGCAT_CAT_RESULT)
+# A mocked piece of adb logcat output.
+MOCK_ADB_LOGCAT = (u'02-29 14:02:19.123 4454 Nothing\n'
+ u'%s'
+ u'02-29 14:02:22.123 4454 Something again and again\n'
+ ) % u''.join(MOCK_ADB_LOGCAT_CAT_RESULT)
+# The expected result of the cat adb operation.
+MOCK_ADB_UNICODE_LOGCAT_CAT_RESULT = [
+ '02-29 14:02:21.456 4454 Something \u901a\n',
+ '02-29 14:02:21.789 4454 Something again\n'
+]
+# A mocked piece of adb logcat output.
+MOCK_ADB_UNICODE_LOGCAT = (
+ u'02-29 14:02:19.123 4454 Nothing\n'
+ u'%s'
+ u'02-29 14:02:22.123 4454 Something again and again\n'
+) % u''.join(MOCK_ADB_UNICODE_LOGCAT_CAT_RESULT)
+
# Mock start and end time of the adb cat.
MOCK_ADB_LOGCAT_BEGIN_TIME = '02-29 14:02:20.123'
MOCK_ADB_LOGCAT_END_TIME = '02-29 14:02:22.000'
@@ -548,8 +561,9 @@
FastbootProxy, MockAdbProxy):
ad = android_device.AndroidDevice(serial='1')
new_log_path = tempfile.mkdtemp()
- with open(os.path.join(new_log_path, 'file.txt'), 'w') as f:
- f.write('hahah.')
+ new_file_path = os.path.join(new_log_path, 'file.txt')
+ with io.open(new_file_path, 'w', encoding='utf-8') as f:
+ f.write(u'hahah.')
expected_msg = '.* Logs already exist .*'
with self.assertRaisesRegex(android_device.Error, expected_msg):
ad.log_path = new_log_path
@@ -625,13 +639,13 @@
utils.create_dir(ad.log_path)
mock_adb_log_path = os.path.join(ad.log_path, 'adblog,%s,%s.txt' %
(ad.model, ad.serial))
- with open(mock_adb_log_path, 'w') as f:
+ with io.open(mock_adb_log_path, 'w', encoding='utf-8') as f:
f.write(MOCK_ADB_LOGCAT)
ad.cat_adb_log('some_test', MOCK_ADB_LOGCAT_BEGIN_TIME)
cat_file_path = os.path.join(
ad.log_path, 'AdbLogExcerpts',
('some_test,02-29 14-02-20.123,%s,%s.txt') % (ad.model, ad.serial))
- with open(cat_file_path, 'r') as f:
+ with io.open(cat_file_path, 'r', encoding='utf-8') as f:
actual_cat = f.read()
self.assertEqual(actual_cat, ''.join(MOCK_ADB_LOGCAT_CAT_RESULT))
# Stops adb logcat.
@@ -639,6 +653,51 @@
@mock.patch(
'mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock_android_device.MockAdbProxy('1'))
+ @mock.patch(
+ 'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ @mock.patch(
+ 'mobly.utils.start_standing_subprocess', return_value='process')
+ @mock.patch('mobly.utils.stop_standing_subprocess')
+ @mock.patch(
+ 'mobly.logger.get_log_line_timestamp',
+ return_value=MOCK_ADB_LOGCAT_END_TIME)
+ def test_AndroidDevice_cat_adb_log_with_unicode(
+ self, mock_timestamp_getter, stop_proc_mock, start_proc_mock,
+ FastbootProxy, MockAdbProxy):
+ """Verifies that AndroidDevice.cat_adb_log loads the correct adb log
+ file, locates the correct adb log lines within the given time range,
+ and writes the lines to the correct output file.
+ """
+ mock_serial = '1'
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ # Direct the log path of the ad to a temp dir to avoid racing.
+ ad._log_path_base = self.tmp_dir
+ # Expect error if attempted to cat adb log before starting adb logcat.
+ expected_msg = ('.* Attempting to cat adb log when none'
+ ' has been collected.')
+ with self.assertRaisesRegex(android_device.Error, expected_msg):
+ ad.cat_adb_log('some_test', MOCK_ADB_LOGCAT_BEGIN_TIME)
+ ad.start_adb_logcat()
+ utils.create_dir(ad.log_path)
+ mock_adb_log_path = os.path.join(ad.log_path, 'adblog,%s,%s.txt' %
+ (ad.model, ad.serial))
+ with io.open(mock_adb_log_path, 'w', encoding='utf-8') as f:
+ f.write(MOCK_ADB_UNICODE_LOGCAT)
+ ad.cat_adb_log('some_test', MOCK_ADB_LOGCAT_BEGIN_TIME)
+ cat_file_path = os.path.join(
+ ad.log_path, 'AdbLogExcerpts',
+ ('some_test,02-29 14-02-20.123,%s,%s.txt') % (ad.model, ad.serial))
+ with io.open(cat_file_path, 'r', encoding='utf-8') as f:
+ actual_cat = f.read()
+ self.assertEqual(actual_cat,
+ ''.join(MOCK_ADB_UNICODE_LOGCAT_CAT_RESULT))
+ # Stops adb logcat.
+ ad.stop_adb_logcat()
+
+ @mock.patch(
+ 'mobly.controllers.android_device_lib.adb.AdbProxy',
return_value=mock.MagicMock())
@mock.patch(
'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
diff --git a/tests/mobly/output_test.py b/tests/mobly/output_test.py
index ff5cae8..d700272 100755
--- a/tests/mobly/output_test.py
+++ b/tests/mobly/output_test.py
@@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+import io
import logging
import mock
import os
@@ -85,7 +86,7 @@
return (summary_file_path, debug_log_path, info_log_path)
def assert_log_contents(self, log_path, whitelist=[], blacklist=[]):
- with open(log_path, 'r') as f:
+ with io.open(log_path, 'r', encoding='utf-8') as f:
content = f.read()
for item in whitelist:
self.assertIn(item, content)
@@ -232,7 +233,7 @@
(summary_file_path, debug_log_path,
info_log_path) = self.assert_output_logs_exist(output_dir)
summary_entries = []
- with open(summary_file_path) as f:
+ with io.open(summary_file_path, 'r', encoding='utf-8') as f:
for entry in yaml.load_all(f):
self.assertTrue(entry['Type'])
summary_entries.append(entry)
@@ -253,7 +254,7 @@
summary_file_path = os.path.join(output_dir,
records.OUTPUT_FILE_SUMMARY)
found = False
- with open(summary_file_path, 'r') as f:
+ with io.open(summary_file_path, 'r', encoding='utf-8') as f:
raw_content = f.read()
f.seek(0)
for entry in yaml.load_all(f):
diff --git a/tests/mobly/records_test.py b/tests/mobly/records_test.py
index fbbc4f4..f818e62 100755
--- a/tests/mobly/records_test.py
+++ b/tests/mobly/records_test.py
@@ -13,9 +13,9 @@
# limitations under the License.
from builtins import str
-from io import open
import copy
+import io
import mock
import os
import shutil
@@ -356,7 +356,7 @@
dump_path = os.path.join(self.tmp_path, 'ha.yaml')
writer = records.TestSummaryWriter(dump_path)
writer.dump(record1.to_dict(), records.TestSummaryEntryType.RECORD)
- with open(dump_path, 'r', encoding='utf-8') as f:
+ with io.open(dump_path, 'r', encoding='utf-8') as f:
content = yaml.load(f)
self.assertEqual(content['Type'],
records.TestSummaryEntryType.RECORD.value)
@@ -375,7 +375,7 @@
dump_path = os.path.join(self.tmp_path, 'ha.yaml')
writer = records.TestSummaryWriter(dump_path)
writer.dump(record1.to_dict(), records.TestSummaryEntryType.RECORD)
- with open(dump_path, 'r', encoding='utf-8') as f:
+ with io.open(dump_path, 'r', encoding='utf-8') as f:
content = yaml.load(f)
self.assertEqual(content['Type'],
records.TestSummaryEntryType.RECORD.value)
@@ -392,7 +392,7 @@
writer = records.TestSummaryWriter(dump_path)
for data in user_data:
writer.dump(data, records.TestSummaryEntryType.USER_DATA)
- with open(dump_path, 'r', encoding='utf-8') as f:
+ with io.open(dump_path, 'r', encoding='utf-8') as f:
contents = []
for c in yaml.load_all(f):
contents.append(c)
diff --git a/tests/mobly/test_runner_test.py b/tests/mobly/test_runner_test.py
index ab26971..c0ff4b5 100755
--- a/tests/mobly/test_runner_test.py
+++ b/tests/mobly/test_runner_test.py
@@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+import io
import logging
import mock
import os
@@ -206,7 +207,7 @@
tr.run()
summary_path = os.path.join(logging.log_path,
records.OUTPUT_FILE_SUMMARY)
- with open(summary_path, 'r') as f:
+ with io.open(summary_path, 'r', encoding='utf-8') as f:
summary_entries = list(yaml.load_all(f))
self.assertEqual(len(summary_entries), 4)
# Verify the first entry is the list of test names.
diff --git a/tests/mobly/utils_test.py b/tests/mobly/utils_test.py
index 7ce4f50..605dc8c 100755
--- a/tests/mobly/utils_test.py
+++ b/tests/mobly/utils_test.py
@@ -12,9 +12,11 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+import io
import mock
import os
import platform
+import shutil
import socket
import subprocess
import tempfile
@@ -36,6 +38,10 @@
def setUp(self):
system = platform.system()
self.sleep_cmd = 'timeout' if system == 'Windows' else 'sleep'
+ self.tmp_dir = tempfile.mkdtemp()
+
+ def tearDown(self):
+ shutil.rmtree(self.tmp_dir)
def test_start_standing_subproc(self):
try:
@@ -61,17 +67,15 @@
self.assertFalse(p1.is_running())
def test_create_dir(self):
- tmp_dir = tempfile.mkdtemp()
- new_path = os.path.join(tmp_dir, 'haha')
+ new_path = os.path.join(self.tmp_dir, 'haha')
self.assertFalse(os.path.exists(new_path))
utils.create_dir(new_path)
self.assertTrue(os.path.exists(new_path))
def test_create_dir_already_exists(self):
- tmp_dir = tempfile.mkdtemp()
- self.assertTrue(os.path.exists(tmp_dir))
- utils.create_dir(tmp_dir)
- self.assertTrue(os.path.exists(tmp_dir))
+ self.assertTrue(os.path.exists(self.tmp_dir))
+ utils.create_dir(self.tmp_dir)
+ self.assertTrue(os.path.exists(self.tmp_dir))
@mock.patch(
'mobly.controllers.android_device_lib.adb.list_occupied_adb_ports')
@@ -114,6 +118,33 @@
finally:
s.close()
+ def test_load_file_to_base64_str_reads_bytes_file_as_base64_string(self):
+ tmp_file_path = os.path.join(self.tmp_dir, 'b64.bin')
+ expected_base64_encoding = u'SGVsbG93IHdvcmxkIQ=='
+ with io.open(tmp_file_path, 'wb') as f:
+ f.write(b'Hellow world!')
+ self.assertEqual(
+ utils.load_file_to_base64_str(tmp_file_path),
+ expected_base64_encoding)
+
+ def test_load_file_to_base64_str_reads_text_file_as_base64_string(self):
+ tmp_file_path = os.path.join(self.tmp_dir, 'b64.bin')
+ expected_base64_encoding = u'SGVsbG93IHdvcmxkIQ=='
+ with io.open(tmp_file_path, 'w', encoding='utf-8') as f:
+ f.write(u'Hellow world!')
+ self.assertEqual(
+ utils.load_file_to_base64_str(tmp_file_path),
+ expected_base64_encoding)
+
+ def test_load_file_to_base64_str_reads_unicode_file_as_base64_string(self):
+ tmp_file_path = os.path.join(self.tmp_dir, 'b64.bin')
+ expected_base64_encoding = u'6YCa'
+ with io.open(tmp_file_path, 'w', encoding='utf-8') as f:
+ f.write(u'\u901a')
+ self.assertEqual(
+ utils.load_file_to_base64_str(tmp_file_path),
+ expected_base64_encoding)
+
if __name__ == '__main__':
unittest.main()
