#437

pending completion

Build # #437

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

<a href="https://github.com/ARMmbed/mbed-os-tools/commit/<a class=hub.com/ARMmbed/mbed-os-tools/commit/<a class="double-link" href="https://git"><a class=hub.com/ARMmbed/mbed-os-tools/commit/<a class="double-link" href="https://git"><a class=hub.com/ARMmbed/mbed-os-tools/commit/<a class="double-link" href="https://git"><a class=hub.com/ARMmbed/mbed-os-tools/commit/<a class="double-link" href="https://git"><a class=hub.com/ARMmbed/mbed-os-tools/commit/835bdbed0964a358cb437ed92790e973939dad27">835bdbed0">&lt;a href=&quot;https://github.com/ARMmbed/mbed-os-tools/commit/</a><a class="double-link" href="https://github.com/ARMmbed/mbed-os-tools/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/ARMmbed/mbed-os-tools/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/ARMmbed/mbed-os-tools/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/ARMmbed/mbed-os-tools/commit/835bdbed0964a358cb437ed92790e973939dad27">835bdbed0</a><a href="https://github.com/ARMmbed/mbed-os-tools/commit/835bdbed0964a358cb437ed92790e973939dad27">&lt;a href=&quot;https://github.com/ARMmbed/mbed-os-tools/commit/835bdbed0964a358cb437ed92790e973939dad27&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/ARMmbed/mbed-os-tools/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/ARMmbed/mbed-os-tools/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/ARMmbed/mbed-os-tools/commit/835bdbed0964a358cb437ed92790e973939dad27&amp;quot;&amp;gt;835bdbed0&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/ARMmbed/mbed-os-tools/commit/835bdbed0964a358cb437ed92790e973939dad27&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/ARMmbed/mbed-os-tools/commit/835bdbed0964a358cb437ed92790e973&amp;lt;/a&amp;gt;939dad27&amp;</a>quot;&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/ARMmbed/mbed-os-tools/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/ARMmbed/mbed-os-tools/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/ARMmbed/mbed-os-to... (continued)

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

3422 of 5710 relevant lines covered (59.93%)

0.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.15

/src/mbed_os_tools/detect/darwin.py

# Copyright (c) 2018, Arm Limited and affiliates.
# SPDX-License-Identifier: Apache-2.0
#
# 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 re
import subprocess
import platform

from bs4 import BeautifulSoup

try:
    from plistlib import loads
except ImportError:
    from plistlib import readPlistFromString as loads
from xml.parsers.expat import ExpatError

from .lstools_base import MbedLsToolsBase

import logging

logger = logging.getLogger("mbedls.lstools_darwin")
logger.addHandler(logging.NullHandler())
DEBUG = logging.DEBUG
del logging

mbed_volume_name_match = re.compile(r"\b(mbed|SEGGER MSD|ATMEL EDBG Media)\b", re.I)


def _plist_from_popen(popen):
    out, _ = popen.communicate()
    if not out:
        return []
    try:
        try:
            # Try simple and fast first if this fails fall back to the slower but
            # more robust process.
            return loads(out)
        except ExpatError:
            # Beautiful soup ensures the XML is properly formed after it is parsed
            # so that it can be used by other less lenient commands without problems
            xml_representation = BeautifulSoup(out.decode('utf8'), 'xml')

            if not xml_representation.get_text():
                # The output is not in the XML format
                return loads(out)
            return loads(xml_representation.decode().encode('utf8'))
    except ExpatError:
        return []


def _find_TTY(obj):
    """ Find the first tty (AKA IODialinDevice) that we can find in the
        children of the specified object, or None if no tty is present.
    """
    try:
        return obj["IODialinDevice"]
    except KeyError:
        for child in obj.get("IORegistryEntryChildren", []):
            found = _find_TTY(child)
            if found:
                return found
        return None


def _prune(current, keys):
    """ Reduce the amount of data we have to sift through to only
        include the specified keys, and children that contain the
        specified keys
    """
    pruned_current = {k: current[k] for k in keys if k in current}
    pruned_children = list(
        filter(
            None, [_prune(c, keys) for c in current.get("IORegistryEntryChildren", [])]
        )
    )
    keep_current = any(k in current for k in keys) or pruned_children
    if keep_current:
        if pruned_children:
            pruned_current["IORegistryEntryChildren"] = pruned_children
        return pruned_current
    else:
        return {}


def _dfs_usb_info(obj, parents):
    """ Find all of the usb info that we can from this particular IORegistry
        tree with depth first search (and searching the parent stack....)
    """
    output = {}
    if (
            "BSD Name" in obj
            and obj["BSD Name"].startswith("disk")
            and mbed_volume_name_match.search(obj["IORegistryEntryName"])
    ):
        disk_id = obj["BSD Name"]
        usb_info = {"serial": None, "vendor_id": None, "product_id": None, "tty": None}
        for parent in [obj] + parents:
            if "USB Serial Number" in parent:
                usb_info["serial"] = parent["USB Serial Number"]
            if "idVendor" in parent and "idProduct" in parent:
                usb_info["vendor_id"] = format(parent["idVendor"], "04x")
                usb_info["product_id"] = format(parent["idProduct"], "04x")
            if usb_info["serial"]:
                usb_info["tty"] = _find_TTY(parent)
            if all(usb_info.values()):
                break
        logger.debug("found usb info %r", usb_info)
        output[disk_id] = usb_info
    for child in obj.get("IORegistryEntryChildren", []):
        output.update(_dfs_usb_info(child, [obj] + parents))
    return output


class MbedLsToolsDarwin(MbedLsToolsBase):
    """ mbed-enabled platform detection on Mac OS X
    """

    def __init__(self, **kwargs):
        MbedLsToolsBase.__init__(self, **kwargs)
        self.mac_version = float(".".join(platform.mac_ver()[0].split(".")[:2]))

    def find_candidates(self):
        # {volume_id: {serial:, vendor_id:, product_id:, tty:}}
        volumes = self._volumes()

        # {volume_id: mount_point}
        mounts = self._mount_points()
        return [
            {
                "mount_point": mounts[v],
                "serial_port": volumes[v]["tty"],
                "target_id_usb_id": volumes[v].get("serial"),
                "vendor_id": volumes[v].get("vendor_id"),
                "product_id": volumes[v].get("product_id"),
            }
            for v in set(volumes.keys()) and set(mounts.keys())
            if v in mounts and v in volumes
        ]

    def _mount_points(self):
        """ Returns map {volume_id: mount_point} """
        diskutil_ls = subprocess.Popen(
            ["diskutil", "list", "-plist"], stdout=subprocess.PIPE
        )
        disks = _plist_from_popen(diskutil_ls)

        if logger.isEnabledFor(DEBUG):
            import pprint

            logger.debug(
                "disks dict \n%s", pprint.PrettyPrinter(indent=2).pformat(disks)
            )
        return {
            disk["DeviceIdentifier"]: disk.get("MountPoint", None)
            for disk in disks["AllDisksAndPartitions"]
        }

    def _volumes(self):
        """ returns a map {volume_id: {serial:, vendor_id:, product_id:, tty:}"""

        # to find all the possible mbed volumes, we look for registry entries
        # under all possible USB tree which have a "BSD Name" that starts with
        # "disk" # (i.e. this is a USB disk), and have a IORegistryEntryName that
        # matches /\cmbed/
        # Once we've found a disk, we can search up for a parent with a valid
        # serial number, and then search down again to find a tty that's part
        # of the same composite device
        # ioreg -a -r -n <usb_controller_name> -l
        usb_controllers = [
            "AppleUSBXHCI",
            "AppleUSBUHCI",
            "AppleUSBEHCI",
            "AppleUSBOHCI",
            "IOUSBHostDevice",
        ]

        cmp_par = "-n"
        # For El Captain we need to list all the instances of (-c) rather than
        # compare names (-n)
        if self.mac_version >= 10.11:
            cmp_par = "-c"

        usb_tree = []
        for usb_controller in usb_controllers:
            ioreg_usb = subprocess.Popen(
                ["ioreg", "-a", "-r", cmp_par, usb_controller, "-l"],
                stdout=subprocess.PIPE,
            )
            usb_tree.extend(_plist_from_popen(ioreg_usb))

        r = {}

        for name, obj in enumerate(usb_tree):
            pruned_obj = _prune(
                obj,
                [
                    "USB Serial Number",
                    "idVendor",
                    "BSD Name",
                    "IORegistryEntryName",
                    "idProduct",
                    "IODialinDevice",
                ],
            )
            if logger.isEnabledFor(DEBUG):
                import pprint

                logger.debug(
                    "finding in \n%s",
                    pprint.PrettyPrinter(indent=2).pformat(pruned_obj),
                )
            r.update(_dfs_usb_info(pruned_obj, []))

        logger.debug("_volumes return %r", r)
        return r

1	# Copyright (c) 2018, Arm Limited and affiliates.
2	# SPDX-License-Identifier: Apache-2.0
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15
16	import re	1✔
17	import subprocess	1✔
18	import platform	1✔
19
20	from bs4 import BeautifulSoup	1✔
21
22	try:	1✔
23	from plistlib import loads	1✔
24	except ImportError:	×
25	from plistlib import readPlistFromString as loads	×
26	from xml.parsers.expat import ExpatError	1✔
27
28	from .lstools_base import MbedLsToolsBase	1✔
29
30	import logging	1✔
31
32	logger = logging.getLogger("mbedls.lstools_darwin")	1✔
33	logger.addHandler(logging.NullHandler())	1✔
34	DEBUG = logging.DEBUG	1✔
35	del logging	1✔
36
37	mbed_volume_name_match = re.compile(r"\b(mbed\|SEGGER MSD\|ATMEL EDBG Media)\b", re.I)	1✔
38
39
40	def _plist_from_popen(popen):	1✔
41	out, _ = popen.communicate()	1✔
42	if not out:	1✔
43	return []	×
44	try:	1✔
45	try:	1✔
46	# Try simple and fast first if this fails fall back to the slower but
47	# more robust process.
48	return loads(out)	1✔
49	except ExpatError:	×
50	# Beautiful soup ensures the XML is properly formed after it is parsed
51	# so that it can be used by other less lenient commands without problems
52	xml_representation = BeautifulSoup(out.decode('utf8'), 'xml')	×
53
54	if not xml_representation.get_text():	×
55	# The output is not in the XML format
56	return loads(out)	×
57	return loads(xml_representation.decode().encode('utf8'))	×
58	except ExpatError:	×
59	return []	×
60
61
62	def _find_TTY(obj):	1✔
63	""" Find the first tty (AKA IODialinDevice) that we can find in the
64	children of the specified object, or None if no tty is present.
65	"""
66	try:	1✔
67	return obj["IODialinDevice"]	1✔
68	except KeyError:	1✔
69	for child in obj.get("IORegistryEntryChildren", []):	1✔
70	found = _find_TTY(child)	1✔
71	if found:	1✔
72	return found	1✔
73	return None	1✔
74
75
76	def _prune(current, keys):	1✔
77	""" Reduce the amount of data we have to sift through to only
78	include the specified keys, and children that contain the
79	specified keys
80	"""
81	pruned_current = {k: current[k] for k in keys if k in current}	1✔
82	pruned_children = list(	1✔
83	filter(
84	None, [_prune(c, keys) for c in current.get("IORegistryEntryChildren", [])]
85	)
86	)
87	keep_current = any(k in current for k in keys) or pruned_children	1✔
88	if keep_current:	1✔
89	if pruned_children:	1✔
90	pruned_current["IORegistryEntryChildren"] = pruned_children	1✔
91	return pruned_current	1✔
92	else:
93	return {}	×
94
95
96	def _dfs_usb_info(obj, parents):	1✔
97	""" Find all of the usb info that we can from this particular IORegistry
98	tree with depth first search (and searching the parent stack....)
99	"""
100	output = {}	1✔
101	if (	1✔
102	"BSD Name" in obj
103	and obj["BSD Name"].startswith("disk")
104	and mbed_volume_name_match.search(obj["IORegistryEntryName"])
105	):
106	disk_id = obj["BSD Name"]	1✔
107	usb_info = {"serial": None, "vendor_id": None, "product_id": None, "tty": None}	1✔
108	for parent in [obj] + parents:	1✔
109	if "USB Serial Number" in parent:	1✔
110	usb_info["serial"] = parent["USB Serial Number"]	1✔
111	if "idVendor" in parent and "idProduct" in parent:	1✔
112	usb_info["vendor_id"] = format(parent["idVendor"], "04x")	1✔
113	usb_info["product_id"] = format(parent["idProduct"], "04x")	1✔
114	if usb_info["serial"]:	1✔
115	usb_info["tty"] = _find_TTY(parent)	1✔
116	if all(usb_info.values()):	1✔
117	break	1✔
118	logger.debug("found usb info %r", usb_info)	1✔
119	output[disk_id] = usb_info	1✔
120	for child in obj.get("IORegistryEntryChildren", []):	1✔
121	output.update(_dfs_usb_info(child, [obj] + parents))	1✔
122	return output	1✔
123
124
125	class MbedLsToolsDarwin(MbedLsToolsBase):	1✔
126	""" mbed-enabled platform detection on Mac OS X
127	"""
128
129	def __init__(self, **kwargs):	1✔
130	MbedLsToolsBase.__init__(self, **kwargs)	1✔
131	self.mac_version = float(".".join(platform.mac_ver()[0].split(".")[:2]))	1✔
132
133	def find_candidates(self):	1✔
134	# {volume_id: {serial:, vendor_id:, product_id:, tty:}}
135	volumes = self._volumes()	1✔
136
137	# {volume_id: mount_point}
138	mounts = self._mount_points()	1✔
139	return [	1✔
140	{
141	"mount_point": mounts[v],
142	"serial_port": volumes[v]["tty"],
143	"target_id_usb_id": volumes[v].get("serial"),
144	"vendor_id": volumes[v].get("vendor_id"),
145	"product_id": volumes[v].get("product_id"),
146	}
147	for v in set(volumes.keys()) and set(mounts.keys())
148	if v in mounts and v in volumes
149	]
150
151	def _mount_points(self):	1✔
152	""" Returns map {volume_id: mount_point} """
153	diskutil_ls = subprocess.Popen(	1✔
154	["diskutil", "list", "-plist"], stdout=subprocess.PIPE
155	)
156	disks = _plist_from_popen(diskutil_ls)	1✔
157
158	if logger.isEnabledFor(DEBUG):	1✔
159	import pprint	×
160
161	logger.debug(	×
162	"disks dict \n%s", pprint.PrettyPrinter(indent=2).pformat(disks)
163	)
164	return {	1✔
165	disk["DeviceIdentifier"]: disk.get("MountPoint", None)
166	for disk in disks["AllDisksAndPartitions"]
167	}
168
169	def _volumes(self):	1✔
170	""" returns a map {volume_id: {serial:, vendor_id:, product_id:, tty:}"""
171
172	# to find all the possible mbed volumes, we look for registry entries
173	# under all possible USB tree which have a "BSD Name" that starts with
174	# "disk" # (i.e. this is a USB disk), and have a IORegistryEntryName that
175	# matches /\cmbed/
176	# Once we've found a disk, we can search up for a parent with a valid
177	# serial number, and then search down again to find a tty that's part
178	# of the same composite device
179	# ioreg -a -r -n <usb_controller_name> -l
180	usb_controllers = [	1✔
181	"AppleUSBXHCI",
182	"AppleUSBUHCI",
183	"AppleUSBEHCI",
184	"AppleUSBOHCI",
185	"IOUSBHostDevice",
186	]
187
188	cmp_par = "-n"	1✔
189	# For El Captain we need to list all the instances of (-c) rather than
190	# compare names (-n)
191	if self.mac_version >= 10.11:	1✔
192	cmp_par = "-c"	1✔
193
194	usb_tree = []	1✔
195	for usb_controller in usb_controllers:	1✔
196	ioreg_usb = subprocess.Popen(	1✔
197	["ioreg", "-a", "-r", cmp_par, usb_controller, "-l"],
198	stdout=subprocess.PIPE,
199	)
200	usb_tree.extend(_plist_from_popen(ioreg_usb))	1✔
201
202	r = {}	1✔
203
204	for name, obj in enumerate(usb_tree):	1✔
205	pruned_obj = _prune(	1✔
206	obj,
207	[
208	"USB Serial Number",
209	"idVendor",
210	"BSD Name",
211	"IORegistryEntryName",
212	"idProduct",
213	"IODialinDevice",
214	],
215	)
216	if logger.isEnabledFor(DEBUG):	1✔
217	import pprint	×
218
219	logger.debug(	×
220	"finding in \n%s",
221	pprint.PrettyPrinter(indent=2).pformat(pruned_obj),
222	)
223	r.update(_dfs_usb_info(pruned_obj, []))	1✔
224
225	logger.debug("_volumes return %r", r)	1✔
226	return r	1✔

ARMmbed / mbed-os-tools / #437

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous