How to use fauxmo - 10 common examples

req = mock.call_args[0][0]
        assert req.full_url == (
            f"http://localhost:8123/api/services/{device.domain}/{signal}"
        )
        assert (
            req.data.decode()
            == f'{{"entity_id": "{device_conf["entity_id"]}"}}'
        )
        assert req.headers == {
            "Authorization": "Bearer abc123",
            "Content-type": "application/json",
        }
        assert int(req.host.rsplit(":")[-1]) == plugin_vars["ha_port"]
        assert req.get_method() == "POST"

        signal = HomeAssistantPlugin.service_map[device.domain.lower()]["off"]
        assert device.off() is True
        req = mock.call_args[0][0]
        assert req.full_url == (
            f"http://localhost:8123/api/services/{device.domain}/{signal}"
        )
        assert (
            req.data.decode()
            == f'{{"entity_id": "{device_conf["entity_id"]}"}}'
        )
        assert req.headers == {
            "Authorization": "Bearer abc123",
            "Content-type": "application/json",
        }
        assert int(req.host.rsplit(":")[-1]) == plugin_vars["ha_port"]
        assert req.get_method() == "POST"

"""Ensure a reasonable HomeAssistant REST endpoint is called."""
    with open(config_path_str) as f:
        config = json.load(f)

    response = MagicMock(spec=HTTPResponse)
    response.status = 200
    response.read.return_value = b'{"state": "off"}'
    mock.return_value.__enter__.return_value = response

    plugin_vars = {
        k: v
        for k, v in config["PLUGINS"]["HomeAssistantPlugin"].items()
        if k not in {"DEVICES", "path"}
    }
    for device_conf in config["PLUGINS"]["HomeAssistantPlugin"]["DEVICES"]:
        device = HomeAssistantPlugin(**{**plugin_vars, **device_conf})

        signal = HomeAssistantPlugin.service_map[device.domain.lower()]["on"]
        assert device.on() is True
        req = mock.call_args[0][0]
        assert req.full_url == (
            f"http://localhost:8123/api/services/{device.domain}/{signal}"
        )
        assert (
            req.data.decode()
            == f'{{"entity_id": "{device_conf["entity_id"]}"}}'
        )
        assert req.headers == {
            "Authorization": "Bearer abc123",
            "Content-type": "application/json",
        }
        assert int(req.host.rsplit(":")[-1]) == plugin_vars["ha_port"]

def test_old_config_fails() -> None:
    """Ensure the config for fauxmo < v0.4.0 fails with SystemExit."""
    with pytest.raises(SystemExit):
        fauxmo.main(config_path_str="tests/old-config-sample.json")

def test_content_length() -> None:
    """Test `CONTENT-LENGTH` HTTP header with non-ascii characters.

    https://github.com/n8henrie/fauxmo/issues/70

    """
    assert "CONTENT-LENGTH: 3" in Fauxmo.add_http_headers("foo")
    assert "CONTENT-LENGTH: 4" in Fauxmo.add_http_headers("föo")

def test_content_length() -> None:
    """Test `CONTENT-LENGTH` HTTP header with non-ascii characters.

    https://github.com/n8henrie/fauxmo/issues/70

    """
    assert "CONTENT-LENGTH: 3" in Fauxmo.add_http_headers("foo")
    assert "CONTENT-LENGTH: 4" in Fauxmo.add_http_headers("föo")

)
        print(errmsg)
        sys.exit(1)

    for plugin in plugins:

        modname = f"{__package__}.plugins.{plugin.lower()}"
        try:
            module = importlib.import_module(modname)

        except ModuleNotFoundError:
            path_str = config["PLUGINS"][plugin]["path"]
            module = module_from_file(modname, path_str)

        PluginClass = getattr(module, plugin)  # noqa
        if not issubclass(PluginClass, FauxmoPlugin):
            raise TypeError(f"Plugins must inherit from {repr(FauxmoPlugin)}")

        # Pass along variables defined at the plugin level that don't change
        # per device
        plugin_vars = {
            k: v
            for k, v in config["PLUGINS"][plugin].items()
            if k not in {"DEVICES", "path"}
        }
        logger.debug(f"plugin_vars: {repr(plugin_vars)}")

        for device in config["PLUGINS"][plugin]["DEVICES"]:
            logger.debug(f"device config: {repr(device)}")

            # Ensure port is `int`, set it if not given (`None`) or 0
            device["port"] = int(device.get("port", 0)) or get_unused_port()

"/eventservice.xml"
            ""
            ""
            "urn:Belkin:service:metainfo:1"
            "urn:Belkin:serviceId:metainfo1"
            "/upnp/control/metainfo1"
            "/upnp/event/metainfo1"
            "/metainfoservice.xml"
            ""
            ""
            ""
            ""
        )

        setup_response = self.add_http_headers(setup_xml)
        logger.debug(f"Fauxmo response to setup request:\n{setup_response}")
        self.transport.write(setup_response.encode())
        self.transport.close()

"""Run the main fauxmo process.

    Spawns a UDP server to handle the Echo's UPnP / SSDP device discovery
    process as well as multiple TCP servers to respond to the Echo's device
    setup requests and handle its process for turning devices on and off.

    Args:
        config_path_str: Path to config file. If not given will search for
                         `config.json` in cwd, `~/.fauxmo/`, and
                         `/etc/fauxmo/`.
        verbosity: Logging verbosity, defaults to 20

    """
    logger.setLevel(verbosity)
    logger.info(f"Fauxmo {__version__}")
    logger.debug(sys.version)

    if config_path_str:
        config_path = pathlib.Path(config_path_str)
    else:
        for config_dir in (".", "~/.fauxmo", "/etc/fauxmo"):
            config_path = pathlib.Path(config_dir).expanduser() / "config.json"
            if config_path.is_file():
                logger.info(f"Using config: {config_path}")
                break

    try:
        config = json.loads(config_path.read_text())
    except FileNotFoundError:
        logger.error(
            "Could not find config file in default search path. Try "
            "specifying your file with `-c`.\n"

def datagram_received(
        self, data: Union[bytes, Text], addr: Tuple[str, int]
    ) -> None:
        """Check incoming UDP data for requests for Wemo devices.

        Args:
            data: Incoming data content
            addr: Address sending data

        """
        if isinstance(data, bytes):
            data = data.decode("utf8")

        logger.debug(f"Received data below from {addr}:")
        logger.debug(data)

        discover_patterns = [
            "ST: urn:Belkin:device:**",
            "ST: upnp:rootdevice",
            "ST: ssdp:all",
        ]

        discover_pattern = next(
            (pattern for pattern in discover_patterns if pattern in data), None
        )
        if (
            'man: "ssdp:discover"' in data.lower()
            and discover_pattern is not None
        ):
            mx = 0.0

def datagram_received(
        self, data: Union[bytes, Text], addr: Tuple[str, int]
    ) -> None:
        """Check incoming UDP data for requests for Wemo devices.

        Args:
            data: Incoming data content
            addr: Address sending data

        """
        if isinstance(data, bytes):
            data = data.decode("utf8")

        logger.debug(f"Received data below from {addr}:")
        logger.debug(data)

        discover_patterns = [
            "ST: urn:Belkin:device:**",
            "ST: upnp:rootdevice",
            "ST: ssdp:all",
        ]

        discover_pattern = next(
            (pattern for pattern in discover_patterns if pattern in data), None
        )
        if (
            'man: "ssdp:discover"' in data.lower()
            and discover_pattern is not None
        ):
            mx = 0.0
            mx_line = next(