Retrieve native application Python code

.gitignore

@@ -14,3 +14,7 @@
 /tags
 /extension
 /pure_tabs.xpi
+
+**/*.egg-info/
+**/__pycache__/
+**/.pdm-python

README.md

@@ -5,3 +5,11 @@ Basically just a stripped out version of : https://github.com/Kazy/PureTabs
 Tests can be run with `spago test` [TODO : for some reason setting npm "test" script to this value won't work]
 
 Javascript source maps won't work out-of-the-box if JS source files are in e.g. `extension/src` instead of `extension`. This can surely be solved using this issue : https://github.com/parcel-bundler/parcel/issues/3750
+
+## Native backend
+
+The backend requires `pdm` for installation.
+
+It can be installed with `native/install.sh`.
+
+Tests can be run with `(cd native && pdm run pytest)`.

native/install.sh

@@ -0,0 +1,31 @@
+#!/bin/sh
+# Install the Python application and the "backend" manifest, used to allow the extension to communicate
+# with a background process.
+#
+# https://developer.mozilla.org/en-US/docs/Mozilla/Add-ons/WebExtensions/Native_manifests
+
+set -eu
+
+EXTENSION_NAME="job_search_background"
+EXTENSION_BACKEND_PATH=$(realpath $0 | xargs dirname)
+EXTENSION_BIN=${EXTENSION_BACKEND_PATH}/run.sh
+
+NATIVE_MESSAGING_DIR=~/.mozilla/native-messaging-hosts
+NATIVE_MESSAGING_MANIFEST=${NATIVE_MESSAGING_DIR}/${EXTENSION_NAME}.json
+
+mkdir -p ${NATIVE_MESSAGING_DIR}
+
+(cd ${EXTENSION_BACKEND_PATH} && pdm sync)
+
+cat << EOF > ${NATIVE_MESSAGING_MANIFEST}
+{
+    "name": "${EXTENSION_NAME}",
+    "description": "Example host for native messaging",
+    "path": "${EXTENSION_BIN}",
+    "type": "stdio",
+    "allowed_extensions": [
+        "job_search@herve.info",
+        "new_ext@mozilla.org"
+    ]
+}
+EOF

native/pdm.lock

@@ -0,0 +1,88 @@
+# This file is @generated by PDM.
+# It is not intended for manual editing.
+
+[metadata]
+groups = ["default", "dev"]
+cross_platform = true
+static_urls = false
+lock_version = "4.3"
+content_hash = "sha256:93e6b1f3ff197cf8033ffee0aeaf027f986e7bf3da66c5eaa4bc7118fc8fc5a1"
+
+[[package]]
+name = "colorama"
+version = "0.4.6"
+requires_python = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7"
+summary = "Cross-platform colored terminal text."
+files = [
+    {file = "colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6"},
+    {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"},
+]
+
+[[package]]
+name = "iniconfig"
+version = "2.0.0"
+requires_python = ">=3.7"
+summary = "brain-dead simple config-ini parsing"
+files = [
+    {file = "iniconfig-2.0.0-py3-none-any.whl", hash = "sha256:b6a85871a79d2e3b22d2d1b94ac2824226a63c6b741c88f7ae975f18b6778374"},
+    {file = "iniconfig-2.0.0.tar.gz", hash = "sha256:2d91e135bf72d31a410b17c16da610a82cb55f6b0477d1a902134b24a455b8b3"},
+]
+
+[[package]]
+name = "packaging"
+version = "23.2"
+requires_python = ">=3.7"
+summary = "Core utilities for Python packages"
+files = [
+    {file = "packaging-23.2-py3-none-any.whl", hash = "sha256:8c491190033a9af7e1d931d0b5dacc2ef47509b34dd0de67ed209b5203fc88c7"},
+    {file = "packaging-23.2.tar.gz", hash = "sha256:048fb0e9405036518eaaf48a55953c750c11e1a1b68e0dd1a9d62ed0c092cfc5"},
+]
+
+[[package]]
+name = "pluggy"
+version = "1.3.0"
+requires_python = ">=3.8"
+summary = "plugin and hook calling mechanisms for python"
+files = [
+    {file = "pluggy-1.3.0-py3-none-any.whl", hash = "sha256:d89c696a773f8bd377d18e5ecda92b7a3793cbe66c87060a6fb58c7b6e1061f7"},
+    {file = "pluggy-1.3.0.tar.gz", hash = "sha256:cf61ae8f126ac6f7c451172cf30e3e43d3ca77615509771b3a984a0730651e12"},
+]
+
+[[package]]
+name = "pytest"
+version = "7.4.3"
+requires_python = ">=3.7"
+summary = "pytest: simple powerful testing with Python"
+dependencies = [
+    "colorama; sys_platform == \"win32\"",
+    "iniconfig",
+    "packaging",
+    "pluggy<2.0,>=0.12",
+]
+files = [
+    {file = "pytest-7.4.3-py3-none-any.whl", hash = "sha256:0d009c083ea859a71b76adf7c1d502e4bc170b80a8ef002da5806527b9591fac"},
+    {file = "pytest-7.4.3.tar.gz", hash = "sha256:d989d136982de4e3b29dabcc838ad581c64e8ed52c11fbe86ddebd9da0818cd5"},
+]
+
+[[package]]
+name = "pyyaml"
+version = "6.0.1"
+requires_python = ">=3.6"
+summary = "YAML parser and emitter for Python"
+files = [
+    {file = "PyYAML-6.0.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:6965a7bc3cf88e5a1c3bd2e0b5c22f8d677dc88a455344035f03399034eb3007"},
+    {file = "PyYAML-6.0.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:f003ed9ad21d6a4713f0a9b5a7a0a79e08dd0f221aff4525a2be4c346ee60aab"},
+    {file = "PyYAML-6.0.1-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:42f8152b8dbc4fe7d96729ec2b99c7097d656dc1213a3229ca5383f973a5ed6d"},
+    {file = "PyYAML-6.0.1-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:062582fca9fabdd2c8b54a3ef1c978d786e0f6b3a1510e0ac93ef59e0ddae2bc"},
+    {file = "PyYAML-6.0.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d2b04aac4d386b172d5b9692e2d2da8de7bfb6c387fa4f801fbf6fb2e6ba4673"},
+    {file = "PyYAML-6.0.1-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:e7d73685e87afe9f3b36c799222440d6cf362062f78be1013661b00c5c6f678b"},
+    {file = "PyYAML-6.0.1-cp311-cp311-win32.whl", hash = "sha256:1635fd110e8d85d55237ab316b5b011de701ea0f29d07611174a1b42f1444741"},
+    {file = "PyYAML-6.0.1-cp311-cp311-win_amd64.whl", hash = "sha256:bf07ee2fef7014951eeb99f56f39c9bb4af143d8aa3c21b1677805985307da34"},
+    {file = "PyYAML-6.0.1-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:855fb52b0dc35af121542a76b9a84f8d1cd886ea97c84703eaa6d88e37a2ad28"},
+    {file = "PyYAML-6.0.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:40df9b996c2b73138957fe23a16a4f0ba614f4c0efce1e9406a184b6d07fa3a9"},
+    {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c22bec3fbe2524cde73d7ada88f6566758a8f7227bfbf93a408a9d86bcc12a0"},
+    {file = "PyYAML-6.0.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:8d4e9c88387b0f5c7d5f281e55304de64cf7f9c0021a3525bd3b1c542da3b0e4"},
+    {file = "PyYAML-6.0.1-cp312-cp312-win32.whl", hash = "sha256:d483d2cdf104e7c9fa60c544d92981f12ad66a457afae824d146093b8c294c54"},
+    {file = "PyYAML-6.0.1-cp312-cp312-win_amd64.whl", hash = "sha256:0d3304d8c0adc42be59c5f8a4d9e3d7379e6955ad754aa9d6ab7a398b59dd1df"},
+    {file = "PyYAML-6.0.1.tar.gz", hash = "sha256:bfdf460b1736c775f2ba9f6a92bca30bc2095067b8a9d77876d1fad6cc3b4a43"},
+]

native/pyproject.toml

@@ -0,0 +1,18 @@
+[project]
+name = "job_search"
+version = "0.1.0"
+description = "Backend for job-search Firefox extension"
+authors = [
+    {name = "jherve", email = "julien.jev.herve@gmail.com"},
+]
+dependencies = [
+    "pyyaml>=6.0.1",
+]
+requires-python = ">=3.11"
+readme = "README.md"
+license = {text = "MIT"}
+
+[tool.pdm.dev-dependencies]
+dev = [
+    "pytest>=7.4.3",
+]

native/run.sh

@@ -0,0 +1,6 @@
+#!/bin/sh
+
+EXTENSION_BACKEND_PATH=$(realpath $0 | xargs dirname)
+PYTHON_BIN=${EXTENSION_BACKEND_PATH}/.venv/bin/python
+
+${PYTHON_BIN} -m job_search.writer

native/src/job_search/job_storage.py

@@ -0,0 +1,238 @@
+import re
+import subprocess
+from pathlib import Path
+from urllib.parse import urlparse, ParseResult
+from dataclasses import dataclass, field, asdict
+from enum import Enum
+from datetime import date, datetime
+from email.utils import parsedate_to_datetime
+
+
+class JobOfferOrigin(Enum):
+    LINKED_IN = "linked_in"
+    OTHER = "other"
+
+
+class CompanyKind(Enum):
+    SSII = "ssii"
+    START_UP = "start_up"
+    HEAD_HUNTER = "head_hunter"
+    REGULAR = "regular"
+
+
+class ApplicationProcess(Enum):
+    LINKED_IN_SIMPLIFIED = "linked_in_simplified"
+    REGULAR = "regular"
+    SPURIOUS = "spurious"
+
+
+class ContractType(Enum):
+    CDI = "CDI"
+    CDD = "CDD"
+    NOT_A_JOB = "not_a_job"
+
+
+class Flexibility(Enum):
+    ON_SITE = "on_site"
+    HYBRID = "hybrid"
+    FULL_REMOTE = "full_remote"
+
+
+def convert_to_parse_result(url):
+    if isinstance(url, str):
+        return urlparse(url)._replace(query=None)
+    elif isinstance(url, ParseResult):
+        return url
+
+
+@dataclass
+class JobOffer:
+    id: str = field(init=False)
+    url: str = field(repr=False)
+    title: str
+    company: str
+    origin: JobOfferOrigin
+    location: str
+    application_process: ApplicationProcess | None = None
+    company_url: str = ""
+    description: str = ""
+    company_kind: CompanyKind | None = None
+    company_domain: str = ""
+    comment: str = ""
+    tags: list[str] = field(default_factory=list)
+    skills: list[str] = field(default_factory=list)
+    publication_date: date = None
+    xp_required: int | None = None
+    first_seen_date: datetime | None = None
+    contract_type: ContractType | None = ContractType.CDI
+    flexibility: Flexibility | None = None
+    alternate_url: str = None
+    _url: ParseResult = field(init=False, repr=False)
+    _company_url: ParseResult = field(init=False, repr=False)
+    _alternate_url: ParseResult = field(init=False, repr=False, default=None)
+
+    def __post_init__(self):
+        self._url = convert_to_parse_result(self.url)
+        self.url = self._url.geturl()
+
+        self._company_url = convert_to_parse_result(self.company_url)
+        self.company_url = self._company_url.geturl()
+
+        if self.alternate_url:
+            self._alternate_url = convert_to_parse_result(self.alternate_url)
+            self.alternate_url = self._alternate_url.geturl()
+
+        if self.origin == JobOfferOrigin.LINKED_IN:
+            path = Path(self._url.path)
+            self.id = f"linked_in_{path.name}"
+
+    def to_storage(self):
+        return {
+            k: v
+            for k, v in asdict(self).items()
+            if k not in ["_url", "_company_url", "_alternate_url"]
+        }
+
+    @staticmethod
+    def from_storage(dict: dict):
+        id = dict.pop("id")
+
+        for field, converter in [
+            ("origin", JobOfferOrigin),
+            ("application_process", ApplicationProcess),
+            ("company_kind", CompanyKind),
+            ("contract_type", ContractType),
+            ("flexibility", Flexibility),
+            ("xp_required", int),
+            ("first_seen_date", parsedate_to_datetime),
+            ("publication_date", date.fromisoformat),
+        ]:
+            if field in dict:
+                dict[field] = converter(dict[field])
+
+        return JobOffer(**dict)
+
+
+def remove_whitespace(s):
+    s = re.sub(r"[^\w\s]", "", s)
+    s = re.sub(r"\s+", "_", s)
+    return s
+
+
+@dataclass
+class JobStorage:
+    base_dir: Path
+    rec_file_path: Path = field(init=False, repr=False)
+
+    def __post_init__(self):
+        if not self.base_dir.is_absolute():
+            raise ValueError(
+                f"The base dir path should be absolute, got '{self.base_dir}'"
+            )
+
+        self.rec_file_path = self.base_dir / "jobs.rec"
+
+        # Create the rec file if it does not exist yet, otherwise
+        # leave it as-is.
+        try:
+            f = open(self.rec_file_path)
+            f.close()
+        except FileNotFoundError:
+            with open(self.rec_file_path, "w+") as f:
+                f.write("%rec: job_offer\n")
+                f.write("%key: id\n")
+                f.write("%type: publication_date date\n")
+                f.write("%type: company_kind enum regular head_hunter ssii start_up\n")
+                f.write(
+                    "%type: application_process enum regular linked_in_simplified\n"
+                )
+                f.write("%type: xp_required range 0 MAX\n")
+                f.write("%type: origin enum linked_in other\n")
+                f.write("%type: contract_type enum CDI CDD not_a_job\n")
+                f.write("%type: flexibility enum on_site hybrid full_remote\n")
+                f.write("%type: first_seen_date date\n")
+                f.write("%auto: first_seen_date\n")
+
+    def read_all(self) -> list[JobOffer]:
+        return {r["id"]: JobOffer.from_storage(r) for r in self.select_all("job_offer")}
+
+    def add_job(self, offer: JobOffer):
+        self.insert(offer)
+
+    def insert(self, offer: JobOffer):
+        self.insert_record("job_offer", offer.to_storage())
+
+    def insert_record(self, type_, fields):
+        cmd_args = [
+            arg
+            for k, v in self.into_args(fields)
+            for arg in ["-f", k, "-v", v]
+            if v is not None and v != [] and v != ""
+        ]
+        cmd = (
+            ["recins", "--verbose", "-t", type_] + cmd_args + [str(self.rec_file_path)]
+        )
+        process = subprocess.run(
+            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True
+        )
+
+        if (code := process.returncode) != 0:
+            error_lines = process.stderr.splitlines()
+            first = error_lines[0]
+
+            if "error: invalid enum value" in first:
+                raise ValueError(f"Found invalid enum value in {fields}")
+            elif "error: duplicated key value in field 'id' in record" in first:
+                raise FileExistsError(f"Duplicate value {fields['id']}")
+            else:
+                raise ValueError(
+                    f"insert command failed with code {code} :\n{process.stderr}"
+                )
+
+    @staticmethod
+    def into_args(fields: dict) -> list[tuple]:
+        args = []
+        for k, v in fields.items():
+            if isinstance(v, list):
+                args += [(k, item) for item in v]
+            elif isinstance(v, int):
+                args += [(k, str(v))]
+            elif isinstance(v, Enum):
+                args += [(k, v.value)]
+            elif isinstance(v, date):
+                args += [(k, v.isoformat())]
+            else:
+                args += [(k, v)]
+        return args
+
+    def select_all(self, type_):
+        cmd = ["recsel", "-t", type_, str(self.rec_file_path)]
+        process = subprocess.run(cmd, stdout=subprocess.PIPE, universal_newlines=True)
+
+        if (code := process.returncode) != 0:
+            raise ValueError(f"select command failed with code {code}")
+
+        dict = {}
+
+        records = []
+        for r in process.stdout.split("\n\n"):
+            dict = {"skills": [], "tags": []}
+
+            lines = re.split(r"\n(?!\+)", r)[:-1]
+            for l in lines:
+                # We assume fields are not empty
+                [field, value] = l.split(": ", 1)
+
+                # Handle multiline records. This will not work if the optional space if not present
+                # after the PLUS sign.
+                value = "\n".join(value.split("\n+ "))
+
+                if field in ["skills", "tags"]:
+                    dict[field].append(value)
+                else:
+                    dict[field] = value
+
+            if lines != []:
+                records.append(dict)
+
+        return records

native/src/job_search/messages.py

@@ -0,0 +1,138 @@
+import re
+from datetime import date
+from dataclasses import dataclass, asdict
+from enum import Enum
+from typing import Optional, Any
+from job_search.job_storage import (
+    JobOffer,
+    ApplicationProcess,
+    JobOfferOrigin,
+    Flexibility,
+)
+
+
+def to_snake_case(string):
+    return "".join("_" + c.lower() if c.isupper() else c for c in string)
+
+
+class Message:
+    ...
+
+
+class BackgroundScriptMessage(Message):
+    @staticmethod
+    def interpret(message):
+        if not isinstance(message, dict):
+            raise TypeError(f"message should be a dict, got {type(message)}")
+
+        try:
+            tag = message.pop("tag")
+        except KeyError:
+            raise ValueError("message should contain a tag")
+
+        message = {to_snake_case(k): v for k, v in message.items()}
+
+        match tag:
+            case "visited_linkedin_job_page":
+                return VisitedLinkedInJobPageMessage(**message)
+            case "initial_configuration":
+                return InitialConfigurationMessage(**message)
+            case _:
+                raise ValueError(f"Got message with unknown tag {tag}")
+
+
+class NativeMessage(Message):
+    def serialize(self):
+        if isinstance(self, JobOfferListMessage):
+            tag = "job_offer_list"
+        elif isinstance(self, JobAddedMessage):
+            tag = "job_added"
+        elif isinstance(self, JobAlreadyExistsMessage):
+            tag = "job_already_exists"
+        elif isinstance(self, LogMessage):
+            tag = "log_message"
+        else:
+            raise TypeError(f"No tag was associated to {type(self)} for serialization")
+
+        return asdict(self) | {"tag": tag}
+
+
+@dataclass
+class VisitedLinkedInJobPageMessage(BackgroundScriptMessage):
+    url: str
+    job_title: str
+    page_title: str
+    company: str
+    location: str
+    has_simplified_process: bool
+    company_url: str
+    flexibility: Optional[str] = None
+    company_domain: Optional[str] = None
+
+    def extract_job_offer(self):
+        application_process = (
+            ApplicationProcess.LINKED_IN_SIMPLIFIED
+            if self.has_simplified_process
+            else ApplicationProcess.REGULAR
+        )
+
+        if isinstance(self.flexibility, str):
+            flexibility = Flexibility(self.flexibility)
+        elif self.flexibility is None:
+            flexibility = None
+
+        return JobOffer(
+            url=self.url,
+            title=self.job_title,
+            company=self.company,
+            origin=JobOfferOrigin.LINKED_IN,
+            application_process=application_process,
+            location=self.location,
+            company_domain=self.company_domain,
+            company_url=self.company_url,
+            flexibility=flexibility,
+        )
+
+
+@dataclass
+class InitialConfigurationMessage(BackgroundScriptMessage):
+    jobs_path: str
+
+
+@dataclass
+class JobOfferListMessage(NativeMessage):
+    job_offers: list[JobOffer]
+
+
+@dataclass
+class JobAddedMessage(NativeMessage):
+    job: JobOffer
+
+
+@dataclass
+class JobAlreadyExistsMessage(NativeMessage):
+    job_id: str
+
+
+class LogLevel(Enum):
+    DEBUG = "debug"
+    INFO = "info"
+    ERROR = "error"
+
+
+@dataclass
+class LogMessage(NativeMessage):
+    level: LogLevel
+    content: Any
+
+    @staticmethod
+    def debug(**kwargs):
+        return LogMessage(level=LogLevel.DEBUG, **kwargs)
+
+    @staticmethod
+    def info(**kwargs):
+        return LogMessage(level=LogLevel.INFO, **kwargs)
+
+    @staticmethod
+    def error(**kwargs):
+        return LogMessage(level=LogLevel.ERROR, **kwargs)

native/src/job_search/read_write.py

@@ -0,0 +1,60 @@
+import json
+import struct
+from enum import Enum
+from dataclasses import dataclass, is_dataclass, asdict
+from datetime import date
+from io import TextIOWrapper
+
+from job_search.messages import Message, BackgroundScriptMessage
+
+
+class EnhancedJSONEncoder(json.JSONEncoder):
+    def default(self, o):
+        if isinstance(o, Message):
+            return o.serialize()
+        elif is_dataclass(o):
+            return asdict(o)
+        elif isinstance(o, Enum):
+            return o.value
+        elif isinstance(o, date):
+            return o.isoformat()
+        return super().default(o)
+
+
+class StdReadWriter:
+    @staticmethod
+    def read_message(stream):
+        raw_length = stream.buffer.read(4)
+        if len(raw_length) == 0:
+            exit(0)
+        message_length = struct.unpack("@I", raw_length)[0]
+        message = stream.buffer.read(message_length).decode("utf-8")
+        return json.loads(message)
+
+    @staticmethod
+    def encode_message(message_content):
+        encoded_content = json.dumps(
+            message_content, separators=(",", ":"), cls=EnhancedJSONEncoder
+        ).encode("utf-8")
+        encoded_length = struct.pack("@I", len(encoded_content))
+        return {"length": encoded_length, "content": encoded_content}
+
+    @staticmethod
+    def send_message_on(stream, encoded_message):
+        stream.buffer.write(encoded_message["length"])
+        stream.buffer.write(encoded_message["content"])
+        stream.buffer.flush()
+
+
+@dataclass
+class ReadWriter:
+    stdin: TextIOWrapper
+    stdout: TextIOWrapper
+
+    def get_message(self):
+        message = StdReadWriter.read_message(self.stdin)
+        return BackgroundScriptMessage.interpret(message)
+
+    def send_message(self, message):
+        encoded = StdReadWriter.encode_message(message)
+        return StdReadWriter.send_message_on(self.stdout, encoded)

native/src/job_search/writer.py

@@ -0,0 +1,56 @@
+import sys
+import traceback
+from pathlib import Path
+
+from job_search.read_write import ReadWriter
+from job_search.job_storage import JobStorage
+from job_search.messages import (
+    VisitedLinkedInJobPageMessage,
+    InitialConfigurationMessage,
+    JobOfferListMessage,
+    LogMessage,
+    Message,
+    JobAddedMessage,
+    JobAlreadyExistsMessage,
+)
+
+
+class Application:
+    def __init__(self, stdin, stdout):
+        self.read_writer = ReadWriter(stdin, stdout)
+        self.job_storage = None
+
+    def handle_message(self, message: Message):
+        match message:
+            case VisitedLinkedInJobPageMessage():
+                offer = message.extract_job_offer()
+
+                try:
+                    self.job_storage.add_job(offer)
+                    self.read_writer.send_message(JobAddedMessage(offer))
+                except FileExistsError as e:
+                    self.read_writer.send_message(JobAlreadyExistsMessage(offer.id))
+
+            case InitialConfigurationMessage(jobs_path):
+                self.job_storage = JobStorage(base_dir=Path(jobs_path))
+
+    def loop(self):
+        while True:
+            try:
+                received_message = self.read_writer.get_message()
+                self.handle_message(received_message)
+
+                if self.job_storage:
+                    self.read_writer.send_message(
+                        JobOfferListMessage(self.job_storage.read_all())
+                    )
+
+            except Exception as e:
+                exc_info = sys.exc_info()
+                tb = "".join(traceback.format_exception(*exc_info))
+                self.read_writer.send_message(LogMessage.error(content=tb))
+
+
+if __name__ == "__main__":
+    app = Application(sys.stdin, sys.stdout)
+    app.loop()

native/tests/test_input.py

@@ -0,0 +1,47 @@
+import pytest
+from datetime import date, datetime
+from job_search.messages import VisitedLinkedInJobPageMessage
+from job_search.job_storage import (
+    JobOffer,
+    JobOfferOrigin,
+    ApplicationProcess,
+    Flexibility,
+)
+
+
+@pytest.fixture(
+    params=[
+        (
+            VisitedLinkedInJobPageMessage(
+                url="https://www.linkedin.com/jobs/view/3755217595",
+                job_title="Job title",
+                page_title="Page title",
+                company="Company",
+                location="location",
+                company_domain="domain",
+                company_url="https://www.linkedin.com/company/the-company/life",
+                has_simplified_process=True,
+                flexibility=Flexibility.FULL_REMOTE.value,
+            ),
+            JobOffer(
+                url="https://www.linkedin.com/jobs/view/3755217595",
+                title="Job title",
+                company="Company",
+                origin=JobOfferOrigin.LINKED_IN,
+                application_process=ApplicationProcess.LINKED_IN_SIMPLIFIED,
+                location="location",
+                company_domain="domain",
+                company_url="https://www.linkedin.com/company/the-company/life",
+                flexibility=Flexibility.FULL_REMOTE,
+            ),
+        ),
+    ]
+)
+def message_job_offer(request):
+    return request.param
+
+
+class TestJobOfferExtraction:
+    def test_extract_from_visited_linkedin(self, message_job_offer):
+        (message, expected_job_offer) = message_job_offer
+        assert message.extract_job_offer() == expected_job_offer

native/tests/test_read_write.py

@@ -0,0 +1,162 @@
+import pytest
+import tempfile
+import os
+from io import TextIOWrapper
+from dataclasses import asdict
+
+from job_search.read_write import StdReadWriter, ReadWriter
+from job_search.messages import (
+    InitialConfigurationMessage,
+    VisitedLinkedInJobPageMessage,
+    JobOfferListMessage,
+    JobAddedMessage,
+    JobAlreadyExistsMessage,
+    LogMessage,
+)
+
+
+def fake_std():
+    (fd, file_path) = tempfile.mkstemp(prefix="job_search")
+    yield file_path
+    os.remove(file_path)
+
+
+@pytest.fixture
+def stdin():
+    yield from fake_std()
+
+
+@pytest.fixture
+def stdout():
+    yield from fake_std()
+
+
+@pytest.fixture
+def stdin_read(stdin):
+    with open(stdin, "rb") as fake:
+        yield TextIOWrapper(fake)
+
+
+@pytest.fixture
+def stdin_write(stdin):
+    with open(stdin, "wb") as fake:
+        yield TextIOWrapper(fake)
+
+
+@pytest.fixture
+def stdout_write(stdout):
+    with open(stdout, "wb") as fake:
+        yield TextIOWrapper(fake)
+
+
+@pytest.fixture
+def stdout_read(stdout):
+    with open(stdout, "rb") as fake:
+        yield TextIOWrapper(fake)
+
+
+@pytest.fixture
+def read_writer(stdin_read, stdout_write):
+    return ReadWriter(stdin_read, stdout_write)
+
+
+class TestStdReadWriter:
+    def test_get_message(self, stdin_read, stdin_write):
+        simple_message = {"test": "pouet"}
+
+        msg = StdReadWriter.encode_message(simple_message)
+        StdReadWriter.send_message_on(stdin_write, msg)
+
+        assert StdReadWriter.read_message(stdin_read) == simple_message
+
+    def test_send(self, stdout_write, stdout_read):
+        simple_message = {"test": "pouet"}
+
+        msg = StdReadWriter.encode_message(simple_message)
+        StdReadWriter.send_message_on(stdout_write, msg)
+
+        assert StdReadWriter.read_message(stdout_read) == simple_message
+
+
+class TestReadWriter:
+    @pytest.fixture(
+        params=[
+            (
+                {"tag": "initial_configuration", "jobsPath": "jobs_path"},
+                InitialConfigurationMessage(jobs_path="jobs_path"),
+            ),
+            (
+                {
+                    "tag": "visited_linkedin_job_page",
+                    "url": "url",
+                    "jobTitle": "job_title",
+                    "pageTitle": "page_title",
+                    "company": "company",
+                    "companyUrl": "company_url",
+                    "companyDomain": "company_domain",
+                    "location": "location",
+                    "hasSimplifiedProcess": True,
+                    "flexibility": "hybrid",
+                },
+                VisitedLinkedInJobPageMessage(
+                    url="url",
+                    job_title="job_title",
+                    page_title="page_title",
+                    company="company",
+                    company_url="company_url",
+                    company_domain="company_domain",
+                    location="location",
+                    has_simplified_process=True,
+                    flexibility="hybrid",
+                ),
+            ),
+        ]
+    )
+    def input_message(self, request):
+        (ext_message_as_dict, message) = request.param
+
+        encoded = StdReadWriter.encode_message(ext_message_as_dict)
+        return message, encoded
+
+    def test_get_message(self, read_writer, stdin_write, input_message):
+        expected_message, encoded = input_message
+        StdReadWriter.send_message_on(stdin_write, encoded)
+
+        assert read_writer.get_message() == expected_message
+
+    @pytest.fixture(
+        params=[
+            (
+                JobOfferListMessage(job_offers=["job_offer_1", "job_offer_2"]),
+                {"tag": "job_offer_list", "job_offers": ["job_offer_1", "job_offer_2"]},
+            ),
+            (
+                JobAddedMessage(job="job"),
+                {"tag": "job_added", "job": "job"},
+            ),
+            (
+                JobAlreadyExistsMessage(job_id="job_id"),
+                {"tag": "job_already_exists", "job_id": "job_id"},
+            ),
+            (
+                LogMessage.debug(content="debug_content"),
+                {"tag": "log_message", "level": "debug", "content": "debug_content"},
+            ),
+            (
+                LogMessage.info(content={"message": "info"}),
+                {"tag": "log_message", "level": "info", "content": {"message": "info"}},
+            ),
+            (
+                LogMessage.error(content="error_content"),
+                {"tag": "log_message", "level": "error", "content": "error_content"},
+            ),
+        ]
+    )
+    def output_message(self, request):
+        return request.param
+
+    def test_send_message(self, read_writer, stdout_read, output_message):
+        original_message, expected_json = output_message
+
+        read_writer.send_message(original_message)
+        assert StdReadWriter.read_message(stdout_read) == expected_json

native/tests/test_storage.py

@@ -0,0 +1,98 @@
+import pytest
+from datetime import date, datetime
+from job_search.job_storage import (
+    JobStorage,
+    JobOffer,
+    JobOfferOrigin,
+    ApplicationProcess,
+    CompanyKind,
+    ContractType,
+    Flexibility,
+)
+
+
+@pytest.fixture(params=[JobStorage])
+def job_storage(request, tmp_path):
+    return request.param(base_dir=tmp_path)
+
+
+@pytest.fixture(
+    params=[
+        JobOffer(
+            url="https://www.linkedin.com/jobs/view/3755217595",
+            title="Job title",
+            company="Company",
+            origin=JobOfferOrigin.LINKED_IN,
+            application_process=ApplicationProcess.REGULAR,
+            location="location",
+            company_domain="domain",
+            company_url="https://www.linkedin.com/company/the-company/life",
+            publication_date=date.today(),
+        ),
+        JobOffer(
+            url="https://www.linkedin.com/jobs/view/3755217595",
+            title="Job title",
+            company="Company",
+            origin=JobOfferOrigin.LINKED_IN,
+            application_process=ApplicationProcess.REGULAR,
+            location="location",
+            company_domain="domain",
+            company_url="https://www.linkedin.com/company/the-company/life",
+            publication_date=date.today(),
+            skills=["skill1", "skill2"],
+            tags=["tag1", "tag2"],
+            xp_required=2,
+            company_kind=CompanyKind.REGULAR,
+            comment="comment",
+            description="description",
+            contract_type=ContractType.CDD,
+            flexibility=Flexibility.HYBRID,
+            alternate_url="https://www.anothersite.com/with/the/offer.html",
+        ),
+        JobOffer(
+            url="https://www.linkedin.com/jobs/view/3755217595",
+            title="Job title",
+            company="Company",
+            origin=JobOfferOrigin.LINKED_IN,
+            application_process=ApplicationProcess.REGULAR,
+            location="location",
+            company_domain="domain",
+            company_url="https://www.linkedin.com/company/the-company/life",
+            comment="""
+multi
+line
+comment
+            """,
+            publication_date=date.today(),
+        ),
+    ]
+)
+def linked_in_job_offer(request):
+    return request.param
+
+
+class TestJobStorage:
+    def test_job_storage_empty_on_startup(self, job_storage):
+        assert job_storage.read_all() == {}
+
+    def test_job_addition(self, job_storage, linked_in_job_offer):
+        job_storage.add_job(linked_in_job_offer)
+
+        all_items = job_storage.read_all().items()
+        assert len(all_items) == 1
+
+        [(id, stored_job)] = all_items
+        assert isinstance(stored_job.first_seen_date, datetime)
+        assert id == stored_job.id
+
+        # Reset the first_seen_date to None, for comparison with the non-stored version
+        stored_job.first_seen_date = None
+        assert stored_job == linked_in_job_offer
+
+    def test_job_duplicate_addition(self, job_storage, linked_in_job_offer):
+        job_storage.add_job(linked_in_job_offer)
+
+        with pytest.raises(FileExistsError) as excinfo:
+            job_storage.add_job(linked_in_job_offer)
+
+        assert linked_in_job_offer.id in str(excinfo.value)

src/Background.purs

@@ -25,7 +25,7 @@ import LinkedIn.UI.Basic.Types (JobFlexibility(..))
 main :: Effect Unit
 main = do
   log "[bg] starting up"
-  port <- connectToNativeApplication "job_search_writer"
+  port <- connectToNativeApplication "job_search_background"
   onNativeMessageAddListener port nativeMessageHandler
   onNativeDisconnectAddListener port \_ -> log "disconnected from native"