* replace black with ruff.

* use unique names for processor methods to help with debugging

.vscode/settings.json

@@ -115,11 +115,18 @@
         "**/[._]*.un~": true,
         "**/_version.py": true,
     },
-    "python.analysis.diagnosticSeverityOverrides": {
-        "reportWildcardImportFromLibrary": "none"
-    },
     "[python]": {
-        "editor.defaultFormatter": "ms-python.black-formatter"
+        "editor.formatOnSave": true,
+        "editor.defaultFormatter": "charliermarsh.ruff"
+    },
+    "python.languageServer": "Pylance",
+    "python.analysis.typeCheckingMode": "basic",
+    "python.analysis.diagnosticSeverityOverrides": {
+        // "reportOptionalSubscript": "none",
+        "reportOptionalMemberAccess": "none",
+        "reportOptionalOperand": "none",
+        "reportPrivateImportUsage": "none"
     },
-    "python.formatting.provider": "none",
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true,
 }

har2locust/__main__.py

@@ -1,14 +1,16 @@
-from argparse import Namespace
+import ast
 import importlib
 import json
 import logging
 import os
 import pathlib
-import ast
 import sys
+from argparse import Namespace
+
 import jinja2
+
 from .argument_parser import get_parser
-from .plugin import entriesprocessor, entriesprocessor_with_args, astprocessor, outputstringprocessor
+from .plugin import astprocessor, entriesprocessor, entriesprocessor_with_args, outputstringprocessor
 
 
 def __main__(arguments=None):
@@ -69,8 +71,10 @@ def process(har: dict, args: Namespace) -> dict:
     values = {"entries": entries}
     for p in entriesprocessor.processors:
         values |= p(entries) or {}
+        # logging.debug(f"{len(entries)} entries after applying {p.__name__}")
     for p in entriesprocessor_with_args.processors:
         values |= p(entries, args) or {}
+        # logging.debug(f"{len(entries)} entries after applying with_args {p.__name__}")
     logging.debug(f"{len(entries)} entries after applying entriesprocessors")
 
     return values

har2locust/argument_parser.py

@@ -1,7 +1,8 @@
-import configargparse
 from argparse import RawDescriptionHelpFormatter
-from ._version import version
 
+import configargparse
+
+from ._version import version
 
 DEFAULT_CONFIG_FILES = ["~/.har2locust.conf", "har2locust.conf"]
 
@@ -11,7 +12,7 @@ def get_parser() -> configargparse.ArgumentParser:
         epilog="""Simplest usage:
 har2locust myrecording.har > locustfile.py
 
-Load extra plugins by import path and/or filename: 
+Load extra plugins by import path and/or filename:
 har2locust --plugins har2locust.extra_plugins.plugin_example,myplugin.py myrecording.har
 
 Disable one of the default plugins:

har2locust/default_plugins/1_resourcefilter.py

@@ -1,9 +1,10 @@
 import logging
+
 from har2locust.plugin import entriesprocessor_with_args
 
 
 @entriesprocessor_with_args
-def process(entries, args):
+def resourcefilter(entries, args):
     resource_types = args.resource_types.split(",")
     known_resource_type = {
         "xhr",

har2locust/default_plugins/default_headers.py

@@ -1,9 +1,11 @@
-from har2locust.plugin import entriesprocessor_with_args
+import logging
 from urllib.parse import urlsplit
 
+from har2locust.plugin import entriesprocessor_with_args
+
 
 @entriesprocessor_with_args  # use with-args version because it is executed last
-def process(entries: list[dict], _args):
+def default_headers(entries: list[dict], _args):
     # calculate headers shared by all requests (same name and value)
     default_headers = None
     for e in entries:

har2locust/default_plugins/headerignore.py

@@ -1,10 +1,11 @@
-from har2locust.plugin import entriesprocessor
-import re
 import pathlib
+import re
+
+from har2locust.plugin import entriesprocessor
 
 
 @entriesprocessor
-def process(entries: list[dict]):
+def headerignore(entries: list[dict]):
     headerignore_path = pathlib.Path(".headerignore")
     filters = []
     if headerignore_path.is_file():

har2locust/default_plugins/rest.py

@@ -1,9 +1,10 @@
-from har2locust.plugin import entriesprocessor
 import logging
 
+from har2locust.plugin import entriesprocessor
+
 
 @entriesprocessor
-def process(entries: list[dict]):
+def rest(entries: list[dict]):
     for e in entries:
         for h in e["response"]["headers"]:
             if h["name"].lower() == "content-type":
@@ -13,7 +14,7 @@ def process(entries: list[dict]):
                         "postData" not in r  # not a post, ok for .rest
                         or r["postData"]["mimeType"] == "application/json"  # json payload, also ok
                     ):
-                        logging.debug(f"{r['url']} is a rest request!")
+                        # logging.debug(f"{r['url']} is a rest request!")
                         r["fname"] = "rest"
                         r["extraparams"] = []  # catch_response=True is already the default for .rest()
 

har2locust/default_plugins/ruff.py

@@ -0,0 +1,27 @@
+import subprocess
+
+from har2locust.plugin import outputstringprocessor
+
+
+@outputstringprocessor
+def ruff(py: str):
+    p = subprocess.Popen(["ruff", "format", "-"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
+    assert p.stdin  # keep linter happy
+    p.stdin.write(py)
+    stdout, _stderr = p.communicate()
+    assert (
+        not p.returncode
+    ), f"Ruff failed to format the output - perhaps your template is broken? unformatted output was: {py}"
+    p = subprocess.Popen(
+        ["ruff", "check", "--fix", "-q", "-"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True
+    )
+    assert p.stdin  # keep linter happy
+    p.stdin.write(stdout)
+    stdout, stderr = p.communicate()
+    # whatever.
+    # assert (
+    #     not p.returncode
+    # ), f"Ruff failed to check/fix the output - perhaps your template is broken? unformatted output was: {py}"
+
+    # for some reason the subprocess returns an extra newline, get rid of that
+    return stdout[:-1]

har2locust/default_plugins/urlignore.py

@@ -1,10 +1,11 @@
-from har2locust.plugin import entriesprocessor
-import re
 import pathlib
+import re
+
+from har2locust.plugin import entriesprocessor
 
 
 @entriesprocessor
-def process(entries: list[dict]):
+def urlignore(entries: list[dict]):
     urlignore_file = pathlib.Path(".urlignore")
     filters = []
     if urlignore_file.is_file():

har2locust/extra_plugins/plugin_example.py

@@ -1,8 +1,13 @@
 # This file has some advanced examples of how to massage your recording
 # Use it as inspiration for the techniques, not as a recommendation for exactly what to do
-from har2locust.plugin import entriesprocessor, astprocessor
-from ast import *
+import ast
+import pathlib
 import re
+import typing
+
+from har2locust.plugin import astprocessor, entriesprocessor
+
+# useful way to debug: print(ast.unparse(node))
 
 
 @entriesprocessor
@@ -14,74 +19,65 @@ def skip_origin_header(entries):
 
 
 @astprocessor
-def get_customer_from_reader(tree: Module, values: dict):
-    class T(NodeTransformer):
-        def visit_ImportFrom(self, node: ImportFrom) -> ImportFrom:
+def inject_authentication(tree: ast.Module, values: dict):
+    class T(ast.NodeTransformer):
+        def visit_ImportFrom(self, node: ast.ImportFrom) -> ast.ImportFrom:
             # our base class is RestUser, not FastHttpUser
             if node.names[0].name == "FastHttpUser":
                 node.module = "svs_locust"
                 node.names[0].name = "RestUser"
             self.generic_visit(node)
             return node
 
-        def visit_ClassDef(self, node: ClassDef) -> ClassDef:
-            node.bases[0].id = "RestUser"
+        def visit_ClassDef(self, node: ast.ClassDef) -> ast.ClassDef:
+            typing.cast(ast.Name, node.bases[0]).id = "RestUser"
             node.body.pop(0)  # remove host
             self.generic_visit(node)
             return node
 
-        def visit_FunctionDef(self, node: FunctionDef) -> FunctionDef:
+        def visit_FunctionDef(self, node: ast.FunctionDef) -> ast.FunctionDef:
             # replace testlogin request with call to self.auth()
             for i in range(len(node.body)):
                 try:
-                    if node.body[i].items[0].context_expr.args[1].value == "/player/1/authenticate/testlogin":  # type: ignore
-                        block = parse(
-                            """
+                    url = node.body[i].items[0].context_expr.args[1].value  # type: ignore
+                except:  # noqa: E722
+                    url = None
+                if url == "/player/1/authenticate/testlogin":
+                    block = ast.parse(
+                        """
 self.customer = next(self.customer_iterator)
 self.auth()
 """
-                        )
-                        node.body[i] = block.body[0]
-                        # yea, the next line modifies what we're iterating over so we'll miss the last element, which is ugly
-                        node.body.insert(i + 1, block.body[1])
-                    # if node.body[i].items[0].context_expr.args[1].value == "/wager/9/wagers":  # type: ignore
-                    #     json_param = [
-                    #         kw_arg.value
-                    #         for kw_arg in node.body[i].items[0].context_expr.keywords  # type: ignore
-                    #         if kw_arg.arg == "json"
-                    #     ][0]
-                    #     bets = [
-                    #         json_param.values[k]
-                    #         for k in range(len(json_param.keys))
-                    #         if json_param.keys[k].value == "bets"
-                    #     ][0]
-                    #     node.body[i] = parse("self.wagerwrapper(game, append_draw_num=True)").body[0]
-                except:
-                    pass
-
-            # Old school
-            #             # wrap the entire task function body in a with-block.
-            #             if node.name == "t":
-            #                 with_block = parse(
-            #                     f"""
-            # with self.reader.user() as self.customer:
-            #     pass
-            #                     """
-            #                 ).body[0]
-            #                 assert isinstance(with_block, With), with_block
-            #                 with_block.body = node.body
-            #                 node.body = [with_block]
+                    )
+                    node.body[i] = block.body[0]
+                    # yea, the next line modifies what we're iterating over so we'll miss the last element, which is ugly
+                    node.body.insert(i + 1, block.body[1])
             self.generic_visit(node)
             return node
 
-        def visit_Call(self, node: Call) -> Call:
+    T().visit(tree)
+
+
+@astprocessor
+def rest_(tree: ast.Module, values: dict):
+    class T(ast.NodeTransformer):
+        def visit_Call(self, node: ast.Call) -> ast.Call:
             # call rest_ instead of rest for those urls that have &_<timestamp> at the end
-            if isinstance(node.func, Attribute) and node.func.attr == "rest":
-                c = node.args[1]
-                if isinstance(c, Constant):
-                    if re.search(r"[&?]_=\d+$", c.value):
-                        node.func.attr = "rest_"
-                        c.value = re.sub(r"[&?]_=\d+$", "", c.value)
+            if isinstance(node.func, ast.Attribute):
+                try:
+                    url = node.args[1]
+                except Exception:
+                    url = None
+                if isinstance(url, ast.Constant):
+                    if node.func.attr == "rest":
+                        if re.search(r"[&?]_=\d+$", url.value):
+                            node.func.attr = "rest_"
+                            url.value = re.sub(r"[&?]_=\d+$", "", url.value)
+            self.generic_visit(node)
+            return node
+
+    T().visit(tree)
+
             self.generic_visit(node)
             return node
 

har2locust/locust.jinja2

@@ -1,6 +1,8 @@
-# note that any comments here will be removed during AST parsing
-from locust import task, run_single_user
+# note that any comments here will be removed during AST parsing, and formatting will remove any unused imports
+from locust import task, run_single_user, events
 from locust import FastHttpUser
+import re
+import os
 
 class {{class_name}}(FastHttpUser):
     host = "{{host}}"

har2locust/plugin.py

@@ -1,6 +1,6 @@
-from typing import Callable, Optional
-from argparse import Namespace
 import ast
+from argparse import Namespace
+from typing import Callable, Optional
 
 # For examples of how to write a plugin, see default_plugins/ or tests/plugin_example.py
 # The processors allow you to interact with your recording at various stages,

pyproject.toml

@@ -6,5 +6,18 @@ build-backend = "setuptools.build_meta"
 write_to = "har2locust/_version.py"
 local_scheme = "no-local-version"
 
-[tool.black]
+[tool.ruff]
+target-version = "py39"
 line-length = 120
+
+[tool.ruff.lint]
+dummy-variable-rgx = "_.*|^ignored_|^unused_|^kwargs|^environment|^resp"
+ignore = ["E402", "E501", "E713", "E731", "E741"]
+select = ["E", "F", "W", "UP", "FA102", "I001"]
+
+[tool.ruff.lint.isort]
+section-order = ["future", "locust", "standard-library", "third-party", "first-party", "local-folder"]
+# Custom selection-order: to ensure locust is imported as first in lucustfiles (for successful gevent monkey patching)
+
+[tool.ruff.lint.isort.sections]
+locust = ["locust"]

setup.cfg

@@ -31,7 +31,7 @@ zip_safe = false
 python_requires = >= 3.9
 install_requires =
     locust >= 2.14.0
-    black
+    ruff
 
 [options.packages.find]
 exclude =

tests/test_har2locust.py

@@ -1,11 +1,13 @@
 # pylint: disable=redefined-outer-name
 import json
+import os
 import pathlib
+import re
 import subprocess
-import os
+
 import pytest
-import re
-from har2locust.__main__ import __main__, process, render, load_plugins
+
+from har2locust.__main__ import __main__, load_plugins, process, render
 from har2locust.argument_parser import get_parser
 
 inputs_dir = pathlib.Path(__file__).parents[0] / "inputs"