1 Commits
v1 ... auth-v1

Author SHA1 Message Date
Miguel Grinberg
7ee1c7eef9 Authentication support 2022-09-24 19:54:26 +01:00
92 changed files with 879 additions and 2126 deletions

View File

@@ -1,26 +0,0 @@
---
name: Bug report
about: Create a report to help us improve
title: ''
labels: ''
assignees: ''
---
**IMPORTANT**: If you have a question, or you are not sure if you have found a bug in this package, then you are in the wrong place. Hit back in your web browser, and then open a GitHub Discussion instead. Likewise, if you are unable to provide the information requested below, open a discussion to troubleshoot your issue.
**Describe the bug**
A clear and concise description of what the bug is. If you are getting errors, please include the complete error message, including the stack trace.
**To Reproduce**
Steps to reproduce the behavior:
1. Go to '...'
2. Click on '....'
3. Scroll down to '....'
4. See error
**Expected behavior**
A clear and concise description of what you expected to happen.
**Additional context**
Add any other context about the problem here.

View File

@@ -1,5 +0,0 @@
blank_issues_enabled: false
contact_links:
- name: GitHub Discussions
url: https://github.com/miguelgrinberg/microdot/discussions
about: Ask questions here.

View File

@@ -1,20 +0,0 @@
---
name: Feature request
about: Suggest an idea for this project
title: ''
labels: ''
assignees: ''
---
**Is your feature request related to a problem? Please describe.**
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
**Describe the solution you'd like**
A clear and concise description of what you want to happen.
**Describe alternatives you've considered**
A clear and concise description of any alternative solutions or features you've considered.
**Additional context**
Add any other context or screenshots about the feature request here.

View File

@@ -11,8 +11,8 @@ jobs:
name: lint
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v3
- uses: actions/checkout@v2
- uses: actions/setup-python@v2
- run: python -m pip install --upgrade pip wheel
- run: pip install tox tox-gh-actions
- run: tox -eflake8
@@ -21,12 +21,12 @@ jobs:
strategy:
matrix:
os: [ubuntu-latest, macos-latest, windows-latest]
python: ['3.7', '3.8', '3.9', '3.10', '3.11']
python: ['3.6', '3.7', '3.8', '3.9', '3.10']
fail-fast: false
runs-on: ${{ matrix.os }}
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v3
- uses: actions/checkout@v2
- uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python }}
- run: python -m pip install --upgrade pip wheel
@@ -36,8 +36,8 @@ jobs:
name: tests-micropython
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v3
- uses: actions/checkout@v2
- uses: actions/setup-python@v2
- run: python -m pip install --upgrade pip wheel
- run: pip install tox tox-gh-actions
- run: tox -eupy
@@ -45,21 +45,18 @@ jobs:
name: coverage
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v3
- uses: actions/checkout@v2
- uses: actions/setup-python@v2
- run: python -m pip install --upgrade pip wheel
- run: pip install tox tox-gh-actions
- run: pip install tox tox-gh-actions codecov
- run: tox
- uses: codecov/codecov-action@v3
with:
files: ./coverage.xml
fail_ci_if_error: true
- run: codecov
benchmark:
name: benchmark
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v3
- uses: actions/checkout@v2
- uses: actions/setup-python@v2
- run: python -m pip install --upgrade pip wheel
- run: pip install tox tox-gh-actions
- run: tox -ebenchmark

5
.gitignore vendored
View File

@@ -103,8 +103,3 @@ venv.bak/
# mypy
.mypy_cache/
# other
*.der
*.pem
*_txt.py

View File

@@ -1,16 +0,0 @@
version: 2
build:
os: ubuntu-22.04
tools:
python: "3.11"
sphinx:
configuration: docs/conf.py
python:
install:
- method: pip
path: .
extra_requirements:
- docs

View File

@@ -1,85 +1,5 @@
# Microdot change log
**Release 1.3.4** - 2023-11-08
- Handle change in `wait_closed()` behavior in Python 3.12 [#177](https://github.com/miguelgrinberg/microdot/issues/177) ([commit](https://github.com/miguelgrinberg/microdot/commit/5550b20cdd347d59e2aa68f6ebf9e9abffaff9fc))
- Added missing request argument in some documentation examples [#163](https://github.com/miguelgrinberg/microdot/issues/163) ([commit](https://github.com/miguelgrinberg/microdot/commit/744548f8dc33a72512b34c4001ee9c6c1edd22ee))
- Fix minor documentation typos [#161](https://github.com/miguelgrinberg/microdot/issues/161) ([commit](https://github.com/miguelgrinberg/microdot/commit/2e4911d10826cbb3914de4a45e495c3be36543fa)) (thanks **Andy Piper**!)
**Release 1.3.3** - 2023-07-16
- Handle query string arguments without value [#149](https://github.com/miguelgrinberg/microdot/issues/149) ([commit](https://github.com/miguelgrinberg/microdot/commit/3554bc91cb1523efa5b66fe3ef173f8e86e8c2a0))
- Support empty responses with ASGI adapter ([commit](https://github.com/miguelgrinberg/microdot/commit/e09e9830f43af41d38775547637558494151a385))
- Added CORS extension to Python package ([commit](https://github.com/miguelgrinberg/microdot/commit/304ca2ef6881fe718126b3e308211e760109d519))
- Document access to WSGI and ASGI attributes [#153](https://github.com/miguelgrinberg/microdot/issues/153) ([commit](https://github.com/miguelgrinberg/microdot/commit/d99df2c4010ab70c60b86ab334d656903e04eb26))
- Upgrade micropython tests to use v1.20 ([commit](https://github.com/miguelgrinberg/microdot/commit/e0f0565551966ee0238a5a1819c78a13639ad704))
**Release 1.3.2** - 2023-06-13
- In ASGI, return headers as strings and not binary [#144](https://github.com/miguelgrinberg/microdot/issues/144) ([commit](https://github.com/miguelgrinberg/microdot/commit/e92310fa55bbffcdcbb33f560e27c3579d7ac451))
- Incorrect import in `static_async.py` example ([commit](https://github.com/miguelgrinberg/microdot/commit/c07a53943508e64baea160748e67efc92e75b036))
**Release 1.3.1** - 2023-05-21
- Support negative numbers for int path components [#137](https://github.com/miguelgrinberg/microdot/issues/137) ([commit](https://github.com/miguelgrinberg/microdot/commit/a0dd7c8ab6d681932324e56ed101aba861a105a0))
- Use a more conservative default for socket timeout [#130](https://github.com/miguelgrinberg/microdot/issues/130) ([commit](https://github.com/miguelgrinberg/microdot/commit/239cf4ff37268a7e2467b93be44fe9f91cee8aee))
- More robust check for socket timeout error code [#106](https://github.com/miguelgrinberg/microdot/issues/106) ([commit](https://github.com/miguelgrinberg/microdot/commit/efec9f14be7b6f3451e4d1d0fe7e528ce6ca74dc))
- WebSocket error when handling PING packet [#129](https://github.com/miguelgrinberg/microdot/issues/129) ([commit](https://github.com/miguelgrinberg/microdot/commit/87cd098f66e24bed6bbad29b1490a129e355bbb3))
- Explicitly set UTF-8 encoding for HTML files in examples [#132](https://github.com/miguelgrinberg/microdot/issues/132) ([commit](https://github.com/miguelgrinberg/microdot/commit/f81de6d9582f4905b9c2735d3c639b92d7e77994))
**Release 1.3.0** - 2023-04-08
- Cross-Origin Resource Sharing (CORS) extension [#45](https://github.com/miguelgrinberg/microdot/issues/45) ([commit](https://github.com/miguelgrinberg/microdot/commit/67798f7dbffb30018ab4b62a9aaa297f63bc9e64))
- Respond to `HEAD` and `OPTIONS` requests ([commit](https://github.com/miguelgrinberg/microdot/commit/6a31f89673518e79fef5659c04e609b7976a5e34))
- Tolerate slightly invalid formats in query strings [#126](https://github.com/miguelgrinberg/microdot/issues/126) ([commit](https://github.com/miguelgrinberg/microdot/commit/a1b061656fa19dae583951596b0f1f0603652a56))
- Support compressed files in `send_file()` [#93](https://github.com/miguelgrinberg/microdot/issues/93) ([commit](https://github.com/miguelgrinberg/microdot/commit/daf1001ec55ab38e6cdfee4931729a3b7506858b))
- Add `max_age` argument to `send_file()` ([commit](https://github.com/miguelgrinberg/microdot/commit/e684ee32d91d3e2ab9569bb5fd342986c010ffeb))
- Add `update()` method to `NoCaseDict` class ([commit](https://github.com/miguelgrinberg/microdot/commit/ea6766cea96b756b36ed777f9c1b6a6680db09ba))
- Set exit code to 1 for failed MicroPython test runs ([commit](https://github.com/miguelgrinberg/microdot/commit/a350e8fd1e55fac12c9e5b909cfa82d880b177ef))
**Release 1.2.4** - 2023-03-03
- One more attempt to correct build issues ([commit](https://github.com/miguelgrinberg/microdot/commit/cb39898829f4edc233ab4e7ba3f7ef3c5c50f196))
**Release 1.2.3** - 2023-03-03
- Corrected a problem with previous build.
**Release 1.2.2** - 2023-03-03
- Add a socket read timeout to abort incomplete requests [#99](https://github.com/miguelgrinberg/microdot/issues/99) ([commit](https://github.com/miguelgrinberg/microdot/commit/d0d358f94a63f8565d6406feff0c6e7418cc7f81))
- More robust timeout handling [#106](https://github.com/miguelgrinberg/microdot/issues/106) ([commit](https://github.com/miguelgrinberg/microdot/commit/4d432a7d6cd88b874a8b825fb62891ed22881f74))
- Add @after_error_handler decorator [#97](https://github.com/miguelgrinberg/microdot/issues/97) ([commit](https://github.com/miguelgrinberg/microdot/commit/fcaeee69052b5681706f65b022e667baeee30d4d))
- Return headers as lowercase byte sequences as required by ASGI ([commit](https://github.com/miguelgrinberg/microdot/commit/ddb3b8f442d3683df04554104edaf8acd9c68148))
- Async example of static file serving ([commit](https://github.com/miguelgrinberg/microdot/commit/680cd9c023352f0ff03d67f1041ea174b7b7385b))
- Fixing broken links to examples in documentation [#101](https://github.com/miguelgrinberg/microdot/issues/101) ([commit](https://github.com/miguelgrinberg/microdot/commit/c00b24c9436e1b8f3d4c9bb6f2adfca988902e91)) (thanks **Eric Welch**!)
- Add scrollbar to documentation's left sidebar ([commit](https://github.com/miguelgrinberg/microdot/commit/2aa90d42451dc64c84efcc4f40a1b6c8d1ef1e8d))
- Documentation typo [#90](https://github.com/miguelgrinberg/microdot/issues/90) ([commit](https://github.com/miguelgrinberg/microdot/commit/81394980234f24aac834faf8e2e8225231e9014b)) (thanks **William Wheeler**!)
- Add CPU timing to benchmark ([commit](https://github.com/miguelgrinberg/microdot/commit/9398c960752f87bc32d7c4349cbf594e5d678e99))
- Upgrade uasyncio release used in tests ([commit](https://github.com/miguelgrinberg/microdot/commit/3d6815119ca1ec989f704f626530f938c857a8e5))
- Update unittest library for MicroPython ([commit](https://github.com/miguelgrinberg/microdot/commit/ecd84ecb7bd3c29d5af96739442b908badeab804))
- New build of micropython for unit tests ([commit](https://github.com/miguelgrinberg/microdot/commit/818f98d9a4e531e01c0f913813425ab2b40c289d))
- Remove 3.6, add 3.11 to builds ([commit](https://github.com/miguelgrinberg/microdot/commit/dd15d90239b73b5fd413515c9cd4ac23f6d42f67))
**Release 1.2.1** - 2022-12-06
- Error handling invokes parent exceptions [#74](https://github.com/miguelgrinberg/microdot/issues/74) ([commit](https://github.com/miguelgrinberg/microdot/commit/24d74fb8483b04e8abe6e303e06f0a310f32700b)) (thanks **Diego Pomares**!)
- Addressed error when deleting a user session in async app [#86](https://github.com/miguelgrinberg/microdot/issues/86) ([commit](https://github.com/miguelgrinberg/microdot/commit/5a589afd5e519e94e84fc1ee69033f2dad51c3ea))
- Add asyncio file upload example ([commit](https://github.com/miguelgrinberg/microdot/commit/c841cbedda40f59a9d87f6895fdf9fd954f854a2))
- New Jinja and uTemplate examples with Bootstrap ([commit](https://github.com/miguelgrinberg/microdot/commit/211ad953aeedb4c7f73fe210424aa173b4dc7fee))
- Fix typos in documentation [#77](https://github.com/miguelgrinberg/microdot/issues/77) ([commit](https://github.com/miguelgrinberg/microdot/commit/4a9b92b800d3fd87110f7bc9f546c10185ee13bc)) (thanks **Diego Pomares**!)
- Add missing exception argument to error handler example in documentation [#73](https://github.com/miguelgrinberg/microdot/issues/73) ([commit](https://github.com/miguelgrinberg/microdot/commit/c443599089f2127d1cb052dfba8a05c1969d65e3)) (thanks **Diego Pomares**!)
**Release 1.2.0** - 2022-09-25
- Use a case insensitive dict for headers ([commit #1](https://github.com/miguelgrinberg/microdot/commit/b0fd6c432371ca5cb10d07ff84c4deed7aa0ce2e) [commit #2](https://github.com/miguelgrinberg/microdot/commit/a8515c97b030f942fa6ca85cbe1772291468fb0d))
- urlencode() helper function ([commit #1](https://github.com/miguelgrinberg/microdot/commit/672512e086384e808489305502e6ebebcc5a888f) [commit #2](https://github.com/miguelgrinberg/microdot/commit/b133dcc34368853ee685396a1bcb50360e807813))
- Added `request.url` attribute with the complete URL of the request ([commit](https://github.com/miguelgrinberg/microdot/commit/1547e861ee28d43d10fe4c4ed1871345d4b81086))
- Do not log HTTPException occurrences ([commit](https://github.com/miguelgrinberg/microdot/commit/cbefb6bf3a3fdcff8b7a8bacad3449be18e46e3b))
- Cache user session for performance ([commit](https://github.com/miguelgrinberg/microdot/commit/01947b101ebe198312c88d73872e3248024918f0))
- File upload example ([commit](https://github.com/miguelgrinberg/microdot/commit/8ebe81c09b604ddc1123e78ad6bc87ceda5f8597))
- Minor documentation styling fixes ([commit](https://github.com/miguelgrinberg/microdot/commit/4f263c63ab7bb1ce0dd48d8e00f3c6891e1bf07e))
**Release 1.1.1** - 2022-09-18
- Make WebSocket internals consistent between TLS and non-TLS [#61](https://github.com/miguelgrinberg/microdot/issues/61) ([commit](https://github.com/miguelgrinberg/microdot/commit/5693b812ceb2c0d51ec3c991adf6894a87e6fcc7))

View File

@@ -1,5 +0,0 @@
include README.md LICENSE tox.ini
recursive-include docs *
recursive-exclude docs/_build *
recursive-include tests *
exclude **/*.pyc

Binary file not shown.

View File

@@ -1,8 +1,3 @@
.py.class, .py.function, .py.method, .py.property {
margin-top: 20px;
}
div.sphinxsidebar {
max-height: 100%;
overflow-y: auto;
}

View File

@@ -52,12 +52,6 @@ API Reference
.. automodule:: microdot_session
:members:
``microdot_cors`` module
------------------------
.. automodule:: microdot_cors
:members:
``microdot_websocket`` module
------------------------------

View File

@@ -23,7 +23,7 @@ Asynchronous Support with Asyncio
| MicroPython: `uasyncio <https://github.com/micropython/micropython/tree/master/extmod/uasyncio>`_
* - Examples
- | `hello_async.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello/hello_async.py>`_
- | `hello_async.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello_async.py>`_
Microdot can be extended to use an asynchronous programming model based on the
``asyncio`` package. When the :class:`Microdot <microdot_asyncio.Microdot>`
@@ -68,8 +68,8 @@ Using the uTemplate Engine
- | `utemplate <https://github.com/pfalcon/utemplate/tree/master/utemplate>`_
* - Examples
- | `hello.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/templates/utemplate/hello.py>`_
| `hello_utemplate_async.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello/hello_utemplate_async.py>`_
- | `hello_utemplate.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello_utemplate.py>`_
| `hello_utemplate_async.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello_utemplate_async.py>`_
The :func:`render_template <microdot_utemplate.render_template>` function is
used to render HTML templates with the uTemplate engine. The first argument is
@@ -110,7 +110,7 @@ Using the Jinja Engine
- | `Jinja2 <https://jinja.palletsprojects.com/>`_
* - Examples
- | `hello.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/templates/jinja/hello.py>`_
- | `hello_jinja.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello_jinja.py>`_
The :func:`render_template <microdot_jinja.render_template>` function is used
to render HTML templates with the Jinja engine. The first argument is the
@@ -137,8 +137,8 @@ subdirectory. This location can be changed with the
.. note::
The Jinja extension is not compatible with MicroPython.
Maintaining Secure User Sessions
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Maintaing Secure User Sessions
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. list-table::
:align: left
@@ -156,7 +156,7 @@ Maintaining Secure User Sessions
`hmac <https://github.com/micropython/micropython-lib/blob/master/python-stdlib/hmac/hmac.py>`_
* - Examples
- | `login.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/sessions/login.py>`_
- | `login.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/login.py>`_
The session extension provides a secure way for the application to maintain
user sessions. The session is stored as a signed cookie in the client's
@@ -208,42 +208,6 @@ Example::
delete_session(req)
return redirect('/')
Cross-Origin Resource Sharing (CORS)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. list-table::
:align: left
* - Compatibility
- | CPython & MicroPython
* - Required Microdot source files
- | `microdot.py <https://github.com/miguelgrinberg/microdot/tree/main/src/microdot.py>`_
| `microdot_cors.py <https://github.com/miguelgrinberg/microdot/tree/main/src/microdot_cors.py>`_
* - Required external dependencies
- | None
* - Examples
- | `cors.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/cors/cors.py>`_
The CORS extension provides support for `Cross-Origin Resource Sharing
(CORS) <https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS>`_. CORS is a
mechanism that allows web applications running on different origins to access
resources from each other. For example, a web application running on
``https://example.com`` can access resources from ``https://api.example.com``.
To enable CORS support, create an instance of the
:class:`CORS <microdot_cors.CORS>` class and configure the desired options.
Example::
from microdot import Microdot
from microdot_cors import CORS
app = Microdot()
cors = CORS(app, allowed_origins=['https://example.com'],
allow_credentials=True)
WebSocket Support
~~~~~~~~~~~~~~~~~
@@ -280,7 +244,7 @@ Example::
ws.send(message)
.. note::
An unsupported *microdot_websocket_alt.py* module, with the same
An unsupported *microsoft_websocket_alt.py* module, with the same
interface, is also provided. This module uses the native WebSocket support
in MicroPython that powers the WebREPL, and may provide slightly better
performance for MicroPython low-end boards. This module is not compatible
@@ -312,9 +276,9 @@ This extension has the same interface as the synchronous WebSocket extension,
but the ``receive()`` and ``send()`` methods are asynchronous.
.. note::
An unsupported *microdot_asgi_websocket.py* module, with the same
An unsupported *microsoft_asgi_websocket.py* module, with the same
interface, is also provided. This module must be used instead of
*microdot_asyncio_websocket.py* when the ASGI support is used. The
*microsoft_asyncio_websocket.py* when the ASGI support is used. The
`echo_asgi.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/websocket/echo_asgi.py>`_
example shows how to use this module.
@@ -333,7 +297,7 @@ HTTPS Support
* - Examples
- | `hello_tls.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/tls/hello_tls.py>`_
| `hello_async_tls.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/tls/hello_async_tls.py>`_
| `hello_asyncio_tls.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/tls/hello_asyncio_tls.py>`_
The ``run()`` function accepts an optional ``ssl`` argument, through which an
initialized ``SSLContext`` object can be passed. MicroPython does not currently
@@ -459,7 +423,7 @@ Using a WSGI Web Server
- | A WSGI web server, such as `Gunicorn <https://gunicorn.org/>`_.
* - Examples
- | `hello_wsgi.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello/hello_wsgi.py>`_
- | `hello_wsgi.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello_wsgi.py>`_
The ``microdot_wsgi`` module provides an extended ``Microdot`` class that
@@ -486,9 +450,6 @@ web application using the Gunicorn web server::
gunicorn test:app
When using this WSGI adapter, the ``environ`` dictionary provided by the web
server is available to request handlers as ``request.environ``.
Using an ASGI Web Server
^^^^^^^^^^^^^^^^^^^^^^^^
@@ -507,7 +468,7 @@ Using an ASGI Web Server
- | An ASGI web server, such as `Uvicorn <https://uvicorn.org/>`_.
* - Examples
- | `hello_asgi.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello/hello_asgi.py>`_
- | `hello_asgi.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello_asgi.py>`_
The ``microdot_asgi`` module provides an extended ``Microdot`` class that
implements the ASGI protocol and can be used with a compliant ASGI server such
@@ -532,5 +493,3 @@ web application using the Uvicorn web server::
uvicorn test:app
When using this ASGI adapter, the ``scope`` dictionary provided by the web
server is available to request handlers as ``request.asgi_scope``.

View File

@@ -283,7 +283,7 @@ handled::
def start_timer(request):
request.g.start_time = time.time()
@app.after_request
@ap.after_request
def end_timer(request, response):
duration = time.time() - request.g.start_time
print(f'Request took {duration:0.2f} seconds')
@@ -293,15 +293,9 @@ The function can return a modified response object to replace the original. If
the function does not return a value, then the original response object is
used.
The after request handlers are only invoked for successful requests. The
:func:`after_error_request() <microdot.Microdot.after_error_request>`
decorator can be used to register a function that is called after an error
occurs. The function receives the request and the error response and is
expected to return an updated response object.
.. note::
The :ref:`request.g <The "g" Object>` object is a special object that allows
the before and after request handlers, as well as the route function to
the before and after request handlers, as well sa the route function to
share data during the life of the request.
Error Handlers
@@ -320,7 +314,7 @@ automatically handled by Microdot are:
While the above errors are fully complaint with the HTTP specification, the
application might want to provide custom responses for them. The
:func:`errorhandler() <microdot.Microdot.errorhandler>` decorator registers
functions to respond to specific error codes. The following example shows a
a functions to respond to specific error codes. The following example shows a
custom error handler for 404 errors::
@app.errorhandler(404)
@@ -328,18 +322,14 @@ custom error handler for 404 errors::
return {'error': 'resource not found'}, 404
The ``errorhandler()`` decorator has a second form, in which it takes an
exception class as an argument. Microdot will then invoke the handler when the
exception is an instance of the given class is raised. The next example
provides a custom response for division by zero errors::
exception class as an argument. Microdot will then invoke the handler when an
exception of that class is raised. The next example provides a custom response
for division by zero errors::
@app.errorhandler(ZeroDivisionError)
def division_by_zero(request, exception):
def division_by_zero(request):
return {'error': 'division by zero'}, 500
When the raised exception class does not have an error handler defined, but
one or more of its base classes do, Microdot makes an attempt to invoke the
most specific handler.
Mounting a Sub-Application
^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -491,7 +481,7 @@ Accessing the Raw Request Body
For cases in which neither JSON nor form data is expected, the
:attr:`body <microdot.Request.body>` request attribute returns the entire body
of the request as a byte sequence.
of the request as a byte sequence.
If the expected body is too large to fit in memory, the application can use the
:attr:`stream <microdot.Request.stream>` request attribute to read the body
@@ -500,7 +490,7 @@ contents as a file-like object.
Cookies
^^^^^^^
Cookies that are sent by the client are made available through the
Cookies that are sent by the client are made available throught the
:attr:`cookies <microdot.Request.cookies>` attribute of the request object in
dictionary form.
@@ -595,7 +585,7 @@ always returned to the client in the response body::
In the above example, Microdot issues a standard 200 status code response, and
inserts the necessary headers.
The application can provide its own status code as a second value returned from
The applicaton can provide its own status code as a second value returned from
the route. The example below returns a 202 status code::
@app.get('/')
@@ -611,7 +601,7 @@ The next example returns an HTML response, instead of a default text response::
return '<h1>Hello, World!</h1>', 202, {'Content-Type': 'text/html'}
If the application needs to return custom headers, but does not need to change
the default status code, then it can return two values, omitting the status
the default status code, then it can return two values, omitting the stauts
code::
@app.get('/')
@@ -655,22 +645,13 @@ File Responses
The :func:`send_file <microdot.Response.send_file>` function builds a response
object for a file::
from microdot import send_file
@app.get('/')
def index(request):
return send_file('/static/index.html')
A suggested caching duration can be returned to the client in the ``max_age``
argument::
from microdot import send_file
@app.get('/')
def image(request):
return send_file('/static/image.jpg', max_age=3600) # in seconds
.. note::
Unlike other web frameworks, Microdot does not automatically configure a
route to serve static files. The following is an example route that can be
@@ -682,7 +663,7 @@ argument::
if '..' in path:
# directory traversal is not allowed
return 'Not found', 404
return send_file('static/' + path, max_age=86400)
return send_file('static/' + path)
Streaming Responses
^^^^^^^^^^^^^^^^^^^
@@ -753,7 +734,7 @@ Another option is to create a response object directly in the route function::
Standard cookies do not offer sufficient privacy and security controls, so
never store sensitive information in them unless you are adding additional
protection mechanisms such as encryption or cryptographic signing. The
:ref:`session <Maintaining Secure User Sessions>` extension implements signed
:ref:`session <Maintaing Secure User Sessions>` extension implements signed
cookies that prevent tampering by malicious actors.
Concurrency

View File

@@ -0,0 +1,27 @@
from microdot import Microdot
from microdot_auth import BasicAuth
app = Microdot()
basic_auth = BasicAuth()
USERS = {
'susan': 'hello',
'david': 'bye',
}
@basic_auth.callback
def verify_password(request, username, password):
if username in USERS and USERS[username] == password:
request.g.user = username
return True
@app.route('/')
@basic_auth
def index(request):
return f'Hello, {request.g.user}!'
if __name__ == '__main__':
app.run(debug=True)

View File

@@ -0,0 +1,60 @@
from microdot import Microdot, redirect
from microdot_session import set_session_secret_key
from microdot_login import LoginAuth
app = Microdot()
set_session_secret_key('top-secret')
login_auth = LoginAuth()
USERS = {
'susan': 'hello',
'david': 'bye',
}
@login_auth.callback
def check_user(request, user_id):
request.g.user = user_id
return True
@app.route('/')
@login_auth
def index(request):
return f'''
<h1>Login Auth Example</h1>
<p>Hello, {request.g.user}!</p>
<form method="POST" action="/logout">
<button type="submit">Logout</button>
</form>
''', {'Content-Type': 'text/html'}
@app.route('/login', methods=['GET', 'POST'])
def login(request):
if request.method == 'GET':
return '''
<h1>Login Auth Example</h1>
<form method="POST">
<input name="username" placeholder="username">
<input name="password" type="password" placeholder="password">
<button type="submit">Login</button>
</form>
''', {'Content-Type': 'text/html'}
username = request.form['username']
password = request.form['password']
if USERS.get(username) == password:
login_auth.login_user(request, username)
return login_auth.redirect_to_next(request)
else:
return redirect('/login')
@app.post('/logout')
def logout(request):
login_auth.logout_user(request)
return redirect('/')
if __name__ == '__main__':
app.run(debug=True)

View File

@@ -0,0 +1,27 @@
from microdot import Microdot
from microdot_auth import TokenAuth
app = Microdot()
token_auth = TokenAuth()
TOKENS = {
'hello': 'susan',
'bye': 'david',
}
@token_auth.callback
def verify_token(request, token):
if token in TOKENS:
request.g.user = TOKENS[token]
return True
@app.route('/')
@token_auth
def index(request):
return f'Hello, {request.g.user}!'
if __name__ == '__main__':
app.run(debug=True)

View File

@@ -1,11 +1,11 @@
aiofiles==0.8.0
anyio==3.6.1
blinker==1.5
certifi==2023.7.22
certifi==2022.6.15
charset-normalizer==2.1.0
click==8.1.3
fastapi==0.79.0
Flask==2.3.2
Flask==2.2.1
gunicorn==20.1.0
h11==0.13.0
h2==4.1.0
@@ -22,12 +22,12 @@ priority==2.0.0
psutil==5.9.1
pydantic==1.9.1
quart==0.18.0
requests==2.31.0
requests==2.28.1
sniffio==1.2.0
starlette==0.27.0
starlette==0.19.1
toml==0.10.2
typing_extensions==4.3.0
urllib3==1.26.18
urllib3==1.26.11
uvicorn==0.18.2
Werkzeug==2.2.3
Werkzeug==2.2.1
wsproto==1.1.0

View File

@@ -0,0 +1,14 @@
curl -X GET http://localhost:5000/ <-- microdot
{"ram": 8429568}%
curl -X GET http://localhost:5000/ <-- microdot_asyncio
{"ram": 12410880}%
curl -X GET http://localhost:8000/ <-- microdot_wsgi
{"ram": 9101312}%
curl -X GET http://localhost:8000/ <-- microdot_asgi
{"ram": 18620416}%
curl -X GET http://localhost:5000/ <-- flask app.run
{"ram":25460736}
curl -X GET http://localhost:5000/ <-- flask run
{"ram":26210304}
curl -X GET http://localhost:5000/ <-- quart run
{"ram":31748096}%

View File

@@ -1,7 +1,6 @@
import os
import subprocess
import time
from timeit import timeit
import requests
import psutil
import humanize
@@ -77,23 +76,19 @@ apps = [
for app, env, name in apps:
p = subprocess.Popen(
app.split() if isinstance(app, str) else app,
env={'PATH': os.environ['PATH'] + ':../../bin', **env},
env={'PATH': os.environ['PATH'], **env},
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
time.sleep(1)
tm = 0
if not name.startswith('baseline'):
def req():
r = requests.get('http://localhost:5000')
r.raise_for_status()
tm = timeit(req, number=1000)
r = requests.get('http://localhost:5000')
r.raise_for_status()
proc = psutil.Process(p.pid)
mem = proc.memory_info().rss
for child in proc.children(recursive=True):
mem += child.memory_info().rss
bar = '*' * (mem // (1024 * 1024))
print(f'{name:<28}{tm:10.2f}s {humanize.naturalsize(mem):>10} {bar}')
print(f'{name:<28}{humanize.naturalsize(mem):>10} {bar}')
p.terminate()
time.sleep(1)

View File

@@ -1 +0,0 @@
This directory contains Cross-Origin Resource Sharing (CORS) examples.

View File

@@ -1,14 +0,0 @@
from microdot import Microdot
from microdot_cors import CORS
app = Microdot()
CORS(app, allowed_origins=['https://example.org'], allow_credentials=True)
@app.route('/')
def index(request):
return 'Hello World!'
if __name__ == '__main__':
app.run()

View File

@@ -2,7 +2,6 @@
<html>
<head>
<title>Microdot GPIO Example</title>
<meta charset="UTF-8">
<link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.3.1/css/bootstrap.min.css" integrity="sha384-ggOyR0iXCbMQv3Xipma34MD+dH/1fQ784/j6cY/iJTQUOhcWr7x9JvoRxT2MZw1T" crossorigin="anonymous">
<script>
function getCookie(name) {

View File

@@ -6,7 +6,6 @@ htmldoc = '''<!DOCTYPE html>
<html>
<head>
<title>Microdot Example Page</title>
<meta charset="UTF-8">
</head>
<body>
<div>

View File

@@ -6,7 +6,6 @@ htmldoc = '''<!DOCTYPE html>
<html>
<head>
<title>Microdot Example Page</title>
<meta charset="UTF-8">
</head>
<body>
<div>

View File

@@ -6,7 +6,6 @@ htmldoc = '''<!DOCTYPE html>
<html>
<head>
<title>Microdot Example Page</title>
<meta charset="UTF-8">
</head>
<body>
<div>

View File

@@ -10,7 +10,7 @@ def index(req):
name = None
if req.method == 'POST':
name = req.form.get('name')
return render_template('index.html', name=name)
return render_template('index_jinja.html', name=name)
if __name__ == '__main__':

View File

@@ -10,7 +10,7 @@ def index(req):
name = None
if req.method == 'POST':
name = req.form.get('name')
return render_template('index.html', name=name)
return render_template('index_utemplate.html', name=name)
if __name__ == '__main__':

View File

@@ -6,7 +6,6 @@ htmldoc = '''<!DOCTYPE html>
<html>
<head>
<title>Microdot Example Page</title>
<meta charset="UTF-8">
</head>
<body>
<div>

View File

@@ -2,7 +2,6 @@
<html>
<head>
<title>Microdot + Jinja example</title>
<meta charset="UTF-8">
</head>
<body>
<h1>Microdot + Jinja example</h1>

View File

@@ -3,7 +3,6 @@
<html>
<head>
<title>Microdot + uTemplate example</title>
<meta charset="UTF-8">
</head>
<body>
<h1>Microdot + uTemplate example</h1>

View File

@@ -6,7 +6,6 @@ BASE_TEMPLATE = '''<!doctype html>
<html>
<head>
<title>Microdot login example</title>
<meta charset="UTF-8">
</head>
<body>
<h1>Microdot login example</h1>
@@ -18,7 +17,7 @@ LOGGED_OUT = '''<p>You are not logged in.</p>
<form method="POST">
<p>
Username:
<input name="username" autofocus />
<input type="text" name="username" autofocus />
</p>
<input type="submit" value="Submit" />
</form>'''

View File

@@ -1,20 +0,0 @@
from microdot import Microdot, send_file
app = Microdot()
@app.route('/')
def index(request):
return send_file('gzstatic/index.html', compressed=True,
file_extension='.gz')
@app.route('/static/<path:path>')
def static(request, path):
if '..' in path:
# directory traversal is not allowed
return 'Not found', 404
return send_file('gzstatic/' + path, compressed=True, file_extension='.gz')
app.run(debug=True)

Binary file not shown.

View File

@@ -2,7 +2,6 @@
<html>
<head>
<title>Static File Serving Demo</title>
<meta charset="UTF-8">
</head>
<body>
<h1>Static File Serving Demo</h1>

View File

@@ -1,18 +0,0 @@
from microdot_asyncio import Microdot, send_file
app = Microdot()
@app.route('/')
async def index(request):
return send_file('static/index.html')
@app.route('/static/<path:path>')
async def static(request, path):
if '..' in path:
# directory traversal is not allowed
return 'Not found', 404
return send_file('static/' + path)
app.run(debug=True)

View File

@@ -19,7 +19,6 @@ def index(request):
<html>
<head>
<title>Microdot Video Streaming</title>
<meta charset="UTF-8">
</head>
<body>
<h1>Microdot Video Streaming</h1>

View File

@@ -21,7 +21,6 @@ def index(request):
<html>
<head>
<title>Microdot Video Streaming</title>
<meta charset="UTF-8">
</head>
<body>
<h1>Microdot Video Streaming</h1>

View File

@@ -1,19 +0,0 @@
from microdot import Microdot, Response
from microdot_jinja import render_template
app = Microdot()
Response.default_content_type = 'text/html'
@app.route('/')
def index(req):
return render_template('page1.html', page='Page 1')
@app.route('/page2')
def page2(req):
return render_template('page2.html', page='Page 2')
if __name__ == '__main__':
app.run()

View File

@@ -1,48 +0,0 @@
<!--
This is based on the Bootstrap 5 starter template from the documentation:
https://getbootstrap.com/docs/5.0/getting-started/introduction/#starter-template
-->
<!doctype html>
<html lang="en">
<head>
<!-- Required meta tags -->
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<!-- Bootstrap CSS -->
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.0.2/dist/css/bootstrap.min.css" rel="stylesheet" integrity="sha384-EVSTQN3/azprG1Anm3QDgpJLIm9Nao0Yz1ztcQTwFspd3yD65VohhpuuCOmLASjC" crossorigin="anonymous">
<title>Microdot + Jinja + Bootstrap</title>
</head>
<body>
<nav class="navbar navbar-expand-lg navbar-light bg-light">
<div class="container">
<a class="navbar-brand" href="/">Microdot + Jinja + Bootstrap</a>
</div>
</nav>
<br>
<div class="container">
<svg xmlns="http://www.w3.org/2000/svg" style="display: none;">
<symbol id="info-fill" fill="currentColor" viewBox="0 0 16 16">
<path d="M8 16A8 8 0 1 0 8 0a8 8 0 0 0 0 16zm.93-9.412-1 4.705c-.07.34.029.533.304.533.194 0 .487-.07.686-.246l-.088.416c-.287.346-.92.598-1.465.598-.703 0-1.002-.422-.808-1.319l.738-3.468c.064-.293.006-.399-.287-.47l-.451-.081.082-.381 2.29-.287zM8 5.5a1 1 0 1 1 0-2 1 1 0 0 1 0 2z"/>
</symbol>
</svg>
<div class="alert alert-primary d-flex align-items-center" role="alert">
<svg class="bi flex-shrink-0 me-2" width="24" height="24" role="img" aria-label="Info:"><use xlink:href="#info-fill"/></svg>
<div>This example demonstrates how to create an application that uses <a href="https://getbootstrap.com" class="alert-link">Bootstrap</a> styling. The page layout is defined in a base template that is inherited by several pages.</div>
</div>
{% block content %}{% endblock %}
</div>
<!-- Optional JavaScript; choose one of the two! -->
<!-- Option 1: Bootstrap Bundle with Popper -->
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.0.2/dist/js/bootstrap.bundle.min.js" integrity="sha384-MrcW6ZMFYlzcLA8Nl+NtUVF0sA7MsXsP1UyJoMp4YLEuNSfAP+JcXn/tWtIaxVXM" crossorigin="anonymous"></script>
<!-- Option 2: Separate Popper and Bootstrap JS -->
<!--
<script src="https://cdn.jsdelivr.net/npm/@popperjs/core@2.9.2/dist/umd/popper.min.js" integrity="sha384-IQsoLXl5PILFhosVNubq5LC7Qb9DXgDA9i+tQ8Zj3iwWAwPtgFTxbJ8NT4GN1R8p" crossorigin="anonymous"></script>
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.0.2/dist/js/bootstrap.min.js" integrity="sha384-cVKIPhGWiC2Al4u+LWgxfKTRIcfu0JTxR+EQDz/bgldoEyl4H0zUF0QKbrJ0EcQF" crossorigin="anonymous"></script>
-->
</body>
</html>

View File

@@ -1,6 +0,0 @@
{% extends 'base.html' %}
{% block content %}
<h2>This is {{ page }}</h2>
<p>Go to <a href="/page2">Page 2</a>.</p>
{% endblock %}

View File

@@ -1,6 +0,0 @@
{% extends 'base.html' %}
{% block content %}
<h2>This is {{ page }}</h2>
<p>Go back <a href="/">Page 1</a>.</p>
{% endblock %}

View File

@@ -1,19 +0,0 @@
from microdot import Microdot, Response
from microdot_utemplate import render_template
app = Microdot()
Response.default_content_type = 'text/html'
@app.route('/')
def index(req):
return render_template('page1.html', page='Page 1')
@app.route('/page2')
def page2(req):
return render_template('page2.html', page='Page 2')
if __name__ == '__main__':
app.run(debug=True)

View File

@@ -1,14 +0,0 @@
</div>
<!-- Optional JavaScript; choose one of the two! -->
<!-- Option 1: Bootstrap Bundle with Popper -->
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.0.2/dist/js/bootstrap.bundle.min.js" integrity="sha384-MrcW6ZMFYlzcLA8Nl+NtUVF0sA7MsXsP1UyJoMp4YLEuNSfAP+JcXn/tWtIaxVXM" crossorigin="anonymous"></script>
<!-- Option 2: Separate Popper and Bootstrap JS -->
<!--
<script src="https://cdn.jsdelivr.net/npm/@popperjs/core@2.9.2/dist/umd/popper.min.js" integrity="sha384-IQsoLXl5PILFhosVNubq5LC7Qb9DXgDA9i+tQ8Zj3iwWAwPtgFTxbJ8NT4GN1R8p" crossorigin="anonymous"></script>
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.0.2/dist/js/bootstrap.min.js" integrity="sha384-cVKIPhGWiC2Al4u+LWgxfKTRIcfu0JTxR+EQDz/bgldoEyl4H0zUF0QKbrJ0EcQF" crossorigin="anonymous"></script>
-->
</body>
</html>

View File

@@ -1,33 +0,0 @@
<!--
This is based on the Bootstrap 5 starter template from the documentation:
https://getbootstrap.com/docs/5.0/getting-started/introduction/#starter-template
-->
<!doctype html>
<html lang="en">
<head>
<!-- Required meta tags -->
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<!-- Bootstrap CSS -->
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.0.2/dist/css/bootstrap.min.css" rel="stylesheet" integrity="sha384-EVSTQN3/azprG1Anm3QDgpJLIm9Nao0Yz1ztcQTwFspd3yD65VohhpuuCOmLASjC" crossorigin="anonymous">
<title>Microdot + uTemplate + Bootstrap</title>
</head>
<body>
<nav class="navbar navbar-expand-lg navbar-light bg-light">
<div class="container">
<a class="navbar-brand" href="/">Microdot + uTemplate + Bootstrap</a>
</div>
</nav>
<br>
<div class="container">
<svg xmlns="http://www.w3.org/2000/svg" style="display: none;">
<symbol id="info-fill" fill="currentColor" viewBox="0 0 16 16">
<path d="M8 16A8 8 0 1 0 8 0a8 8 0 0 0 0 16zm.93-9.412-1 4.705c-.07.34.029.533.304.533.194 0 .487-.07.686-.246l-.088.416c-.287.346-.92.598-1.465.598-.703 0-1.002-.422-.808-1.319l.738-3.468c.064-.293.006-.399-.287-.47l-.451-.081.082-.381 2.29-.287zM8 5.5a1 1 0 1 1 0-2 1 1 0 0 1 0 2z"/>
</symbol>
</svg>
<div class="alert alert-primary d-flex align-items-center" role="alert">
<svg class="bi flex-shrink-0 me-2" width="24" height="24" role="img" aria-label="Info:"><use xlink:href="#info-fill"/></svg>
<div>This example demonstrates how to create an application that uses <a href="https://getbootstrap.com" class="alert-link">Bootstrap</a> styling. The page layout is defined in a base template that is inherited by several pages.</div>
</div>

View File

@@ -1,7 +0,0 @@
{% args page %}
{% include 'base_header.html' %}
<h2>This is {{ page }}</h2>
<p>Go to <a href="/page2">Page 2</a>.</p>
{% include 'base_footer.html' %}

View File

@@ -1,7 +0,0 @@
{% args page %}
{% include 'base_header.html' %}
<h2>This is {{ page }}</h2>
<p>Go back <a href="/">Page 1</a>.</p>
{% include 'base_footer.html' %}

View File

@@ -7,7 +7,6 @@ htmldoc = '''<!DOCTYPE html>
<html>
<head>
<title>Microdot Example Page</title>
<meta charset="UTF-8">
</head>
<body>
<div>

View File

@@ -8,7 +8,6 @@ htmldoc = '''<!DOCTYPE html>
<html>
<head>
<title>Microdot Example Page</title>
<meta charset="UTF-8">
</head>
<body>
<div>

View File

@@ -2,7 +2,6 @@
<html>
<head>
<title>Microdot TLS WebSocket Demo</title>
<meta charset="UTF-8">
</head>
<body>
<h1>Microdot TLS WebSocket Demo</h1>

View File

@@ -2,7 +2,6 @@
<html>
<head>
<title>Microdot Upload Example</title>
<meta charset="UTF-8">
</head>
<body>
<h1>Microdot Upload Example</h1>

View File

@@ -1,7 +1,6 @@
from microdot import Microdot, send_file, Request
from microdot import Microdot, send_file
app = Microdot()
Request.max_content_length = 1024 * 1024 # 1MB (change as needed)
@app.get('/')
@@ -31,4 +30,4 @@ def upload(request):
if __name__ == '__main__':
app.run(debug=True)
app.run()

View File

@@ -1,34 +0,0 @@
from microdot_asyncio import Microdot, send_file, Request
app = Microdot()
Request.max_content_length = 1024 * 1024 # 1MB (change as needed)
@app.get('/')
async def index(request):
return send_file('index.html')
@app.post('/upload')
async def upload(request):
# obtain the filename and size from request headers
filename = request.headers['Content-Disposition'].split(
'filename=')[1].strip('"')
size = int(request.headers['Content-Length'])
# sanitize the filename
filename = filename.replace('/', '_')
# write the file to the files directory in 1K chunks
with open('files/' + filename, 'wb') as f:
while size > 0:
chunk = await request.stream.read(min(size, 1024))
f.write(chunk)
size -= len(chunk)
print('Successfully saved file: ' + filename)
return ''
if __name__ == '__main__':
app.run(debug=True)

View File

@@ -2,7 +2,6 @@
<html>
<head>
<title>Microdot WebSocket Demo</title>
<meta charset="UTF-8">
</head>
<body>
<h1>Microdot WebSocket Demo</h1>

View File

@@ -41,7 +41,7 @@ class SingletonGenerator:
def __next__(self):
if self.state is not None:
_task_queue.push(cur_task, self.state)
_task_queue.push_sorted(cur_task, self.state)
self.state = None
return None
else:
@@ -115,11 +115,11 @@ class IOQueue:
# print('poll', s, sm, ev)
if ev & ~select.POLLOUT and sm[0] is not None:
# POLLIN or error
_task_queue.push(sm[0])
_task_queue.push_head(sm[0])
sm[0] = None
if ev & ~select.POLLIN and sm[1] is not None:
# POLLOUT or error
_task_queue.push(sm[1])
_task_queue.push_head(sm[1])
sm[1] = None
if sm[0] is None and sm[1] is None:
self._dequeue(s)
@@ -142,7 +142,7 @@ def create_task(coro):
if not hasattr(coro, "send"):
raise TypeError("coroutine expected")
t = Task(coro, globals())
_task_queue.push(t)
_task_queue.push_head(t)
return t
@@ -167,7 +167,7 @@ def run_until_complete(main_task=None):
_io_queue.wait_io_event(dt)
# Get next task to run and continue it
t = _task_queue.pop()
t = _task_queue.pop_head()
cur_task = t
try:
# Continue running the coroutine, it's responsible for rescheduling itself
@@ -175,10 +175,6 @@ def run_until_complete(main_task=None):
if not exc:
t.coro.send(None)
else:
# If the task is finished and on the run queue and gets here, then it
# had an exception and was not await'ed on. Throwing into it now will
# raise StopIteration and the code below will catch this and run the
# call_exception_handler function.
t.data = None
t.coro.throw(exc)
except excs_all as er:
@@ -189,37 +185,22 @@ def run_until_complete(main_task=None):
if isinstance(er, StopIteration):
return er.value
raise er
if t.state:
# Task was running but is now finished.
waiting = False
if t.state is True:
# "None" indicates that the task is complete and not await'ed on (yet).
t.state = None
elif callable(t.state):
# The task has a callback registered to be called on completion.
t.state(t, er)
t.state = False
# Schedule any other tasks waiting on the completion of this task
waiting = False
if hasattr(t, "waiting"):
while t.waiting.peek():
_task_queue.push_head(t.waiting.pop_head())
waiting = True
else:
# Schedule any other tasks waiting on the completion of this task.
while t.state.peek():
_task_queue.push(t.state.pop())
waiting = True
# "False" indicates that the task is complete and has been await'ed on.
t.state = False
if not waiting and not isinstance(er, excs_stop):
# An exception ended this detached task, so queue it for later
# execution to handle the uncaught exception if no other task retrieves
# the exception in the meantime (this is handled by Task.throw).
_task_queue.push(t)
# Save return value of coro to pass up to caller.
t.data = er
elif t.state is None:
# Task is already finished and nothing await'ed on the task,
# so call the exception handler.
_exc_context["exception"] = exc
_exc_context["future"] = t
Loop.call_exception_handler(_exc_context)
t.waiting = None # Free waiting queue head
if not waiting and not isinstance(er, excs_stop):
# An exception ended this detached task, so queue it for later
# execution to handle the uncaught exception if no other task retrieves
# the exception in the meantime (this is handled by Task.throw).
_task_queue.push_head(t)
# Indicate task is done by setting coro to the task object itself
t.coro = t
# Save return value of coro to pass up to caller
t.data = er
# Create a new task from a coroutine and run it until it finishes
@@ -256,7 +237,7 @@ class Loop:
def stop():
global _stop_task
if _stop_task is not None:
_task_queue.push(_stop_task)
_task_queue.push_head(_stop_task)
# If stop() is called again, do nothing
_stop_task = None

View File

@@ -17,7 +17,7 @@ class Event:
# Note: This must not be called from anything except the thread running
# the asyncio loop (i.e. neither hard or soft IRQ, or a different thread).
while self.waiting.peek():
core._task_queue.push(self.waiting.pop())
core._task_queue.push_head(self.waiting.pop_head())
self.state = True
def clear(self):
@@ -26,7 +26,7 @@ class Event:
async def wait(self):
if not self.state:
# Event not set, put the calling task on the event's waiting queue
self.waiting.push(core.cur_task)
self.waiting.push_head(core.cur_task)
# Set calling task's data to the event's queue so it can be removed if needed
core.cur_task.data = self.waiting
yield
@@ -36,29 +36,27 @@ class Event:
# MicroPython-extension: This can be set from outside the asyncio event loop,
# such as other threads, IRQs or scheduler context. Implementation is a stream
# that asyncio will poll until a flag is set.
# Note: Unlike Event, this is self-clearing after a wait().
# Note: Unlike Event, this is self-clearing.
try:
import uio
class ThreadSafeFlag(uio.IOBase):
def __init__(self):
self.state = 0
self._flag = 0
def ioctl(self, req, flags):
if req == 3: # MP_STREAM_POLL
return self.state * flags
return self._flag * flags
return None
def set(self):
self.state = 1
def clear(self):
self.state = 0
self._flag = 1
async def wait(self):
if not self.state:
if not self._flag:
yield core._io_queue.queue_read(self)
self.state = 0
self._flag = 0
except ImportError:
pass

View File

@@ -1,51 +1,49 @@
# MicroPython uasyncio module
# MIT license; Copyright (c) 2019-2022 Damien P. George
# MIT license; Copyright (c) 2019-2020 Damien P. George
from . import core
def _run(waiter, aw):
try:
result = await aw
status = True
except BaseException as er:
result = None
status = er
if waiter.data is None:
# The waiter is still waiting, cancel it.
if waiter.cancel():
# Waiter was cancelled by us, change its CancelledError to an instance of
# CancelledError that contains the status and result of waiting on aw.
# If the wait_for task subsequently gets cancelled externally then this
# instance will be reset to a CancelledError instance without arguments.
waiter.data = core.CancelledError(status, result)
async def wait_for(aw, timeout, sleep=core.sleep):
aw = core._promote_to_task(aw)
if timeout is None:
return await aw
def runner(waiter, aw):
nonlocal status, result
try:
result = await aw
s = True
except BaseException as er:
s = er
if status is None:
# The waiter is still waiting, set status for it and cancel it.
status = s
waiter.cancel()
# Run aw in a separate runner task that manages its exceptions.
runner_task = core.create_task(_run(core.cur_task, aw))
status = None
result = None
runner_task = core.create_task(runner(core.cur_task, aw))
try:
# Wait for the timeout to elapse.
await sleep(timeout)
except core.CancelledError as er:
status = er.value
if status is None:
if status is True:
# aw completed successfully and cancelled the sleep, so return aw's result.
return result
elif status is None:
# This wait_for was cancelled externally, so cancel aw and re-raise.
status = True
runner_task.cancel()
raise er
elif status is True:
# aw completed successfully and cancelled the sleep, so return aw's result.
return er.args[1]
else:
# aw raised an exception, propagate it out to the caller.
raise status
# The sleep finished before aw, so cancel aw and raise TimeoutError.
status = True
runner_task.cancel()
await runner_task
raise core.TimeoutError
@@ -55,75 +53,22 @@ def wait_for_ms(aw, timeout):
return wait_for(aw, timeout, core.sleep_ms)
class _Remove:
@staticmethod
def remove(t):
pass
async def gather(*aws, return_exceptions=False):
if not aws:
return []
def done(t, er):
# Sub-task "t" has finished, with exception "er".
nonlocal state
if gather_task.data is not _Remove:
# The main gather task has already been scheduled, so do nothing.
# This happens if another sub-task already raised an exception and
# woke the main gather task (via this done function), or if the main
# gather task was cancelled externally.
return
elif not return_exceptions and not isinstance(er, StopIteration):
# A sub-task raised an exception, indicate that to the gather task.
state = er
else:
state -= 1
if state:
# Still some sub-tasks running.
return
# Gather waiting is done, schedule the main gather task.
core._task_queue.push(gather_task)
ts = [core._promote_to_task(aw) for aw in aws]
for i in range(len(ts)):
if ts[i].state is not True:
# Task is not running, gather not currently supported for this case.
raise RuntimeError("can't gather")
# Register the callback to call when the task is done.
ts[i].state = done
# Set the state for execution of the gather.
gather_task = core.cur_task
state = len(ts)
cancel_all = False
# Wait for the a sub-task to need attention.
gather_task.data = _Remove
try:
yield
except core.CancelledError as er:
cancel_all = True
state = er
# Clean up tasks.
for i in range(len(ts)):
if ts[i].state is done:
# Sub-task is still running, deregister the callback and cancel if needed.
ts[i].state = True
if cancel_all:
ts[i].cancel()
elif isinstance(ts[i].data, StopIteration):
# Sub-task ran to completion, get its return value.
ts[i] = ts[i].data.value
else:
# Sub-task had an exception with return_exceptions==True, so get its exception.
ts[i] = ts[i].data
# Either this gather was cancelled, or one of the sub-tasks raised an exception with
# return_exceptions==False, so reraise the exception here.
if state is not 0:
raise state
# Return the list of return values of each sub-task.
try:
# TODO handle cancel of gather itself
# if ts[i].coro:
# iter(ts[i]).waiting.push_head(cur_task)
# try:
# yield
# except CancelledError as er:
# # cancel all waiting tasks
# raise er
ts[i] = await ts[i]
except Exception as er:
if return_exceptions:
ts[i] = er
else:
raise er
return ts

View File

@@ -22,8 +22,8 @@ class Lock:
raise RuntimeError("Lock not acquired")
if self.waiting.peek():
# Task(s) waiting on lock, schedule next Task
self.state = self.waiting.pop()
core._task_queue.push(self.state)
self.state = self.waiting.pop_head()
core._task_queue.push_head(self.state)
else:
# No Task waiting so unlock
self.state = 0
@@ -31,7 +31,7 @@ class Lock:
async def acquire(self):
if self.state != 0:
# Lock unavailable, put the calling Task on the waiting queue
self.waiting.push(core.cur_task)
self.waiting.push_head(core.cur_task)
# Set calling task's data to the lock's queue so it can be removed if needed
core.cur_task.data = self.waiting
try:

View File

@@ -1,15 +1,13 @@
# This list of package files doesn't include task.py because that's provided
# by the C module.
package(
"uasyncio",
# This list of frozen files doesn't include task.py because that's provided by the C module.
freeze(
"..",
(
"__init__.py",
"core.py",
"event.py",
"funcs.py",
"lock.py",
"stream.py",
"uasyncio/__init__.py",
"uasyncio/core.py",
"uasyncio/event.py",
"uasyncio/funcs.py",
"uasyncio/lock.py",
"uasyncio/stream.py",
),
base_path="..",
opt=3,
)

View File

@@ -26,21 +26,9 @@ class Stream:
# TODO yield?
self.s.close()
async def read(self, n=-1):
r = b""
while True:
yield core._io_queue.queue_read(self.s)
r2 = self.s.read(n)
if r2 is not None:
if n >= 0:
return r2
if not len(r2):
return r
r += r2
async def readinto(self, buf):
async def read(self, n):
yield core._io_queue.queue_read(self.s)
return self.s.readinto(buf)
return self.s.read(n)
async def readexactly(self, n):
r = b""
@@ -64,19 +52,9 @@ class Stream:
return l
def write(self, buf):
if not self.out_buf:
# Try to write immediately to the underlying stream.
ret = self.s.write(buf)
if ret == len(buf):
return
if ret is not None:
buf = buf[ret:]
self.out_buf += buf
async def drain(self):
if not self.out_buf:
# Drain must always yield, so a tight loop of write+drain can't block the scheduler.
return await core.sleep_ms(0)
mv = memoryview(self.out_buf)
off = 0
while off < len(mv):
@@ -97,8 +75,8 @@ async def open_connection(host, port):
from uerrno import EINPROGRESS
import usocket as socket
ai = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)[0] # TODO this is blocking!
s = socket.socket(ai[0], ai[1], ai[2])
ai = socket.getaddrinfo(host, port)[0] # TODO this is blocking!
s = socket.socket()
s.setblocking(False)
ss = Stream(s)
try:
@@ -125,7 +103,16 @@ class Server:
async def wait_closed(self):
await self.task
async def _serve(self, s, cb):
async def _serve(self, cb, host, port, backlog):
import usocket as socket
ai = socket.getaddrinfo(host, port)[0] # TODO this is blocking!
s = socket.socket()
s.setblocking(False)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(ai[-1])
s.listen(backlog)
self.task = core.cur_task
# Accept incoming connections
while True:
try:
@@ -147,20 +134,9 @@ class Server:
# Helper function to start a TCP stream server, running as a new task
# TODO could use an accept-callback on socket read activity instead of creating a task
async def start_server(cb, host, port, backlog=5):
import usocket as socket
# Create and bind server socket.
host = socket.getaddrinfo(host, port)[0] # TODO this is blocking!
s = socket.socket()
s.setblocking(False)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(host[-1])
s.listen(backlog)
# Create and return server object and task.
srv = Server()
srv.task = core.create_task(srv._serve(s, cb))
return srv
s = Server()
core.create_task(s._serve(cb, host, port, backlog))
return s
################################################################################

View File

@@ -99,18 +99,19 @@ class TaskQueue:
def peek(self):
return self.heap
def push(self, v, key=None):
assert v.ph_child is None
assert v.ph_next is None
def push_sorted(self, v, key):
v.data = None
v.ph_key = key if key is not None else core.ticks()
v.ph_key = key
v.ph_child = None
v.ph_next = None
self.heap = ph_meld(v, self.heap)
def pop(self):
def push_head(self, v):
self.push_sorted(v, core.ticks())
def pop_head(self):
v = self.heap
assert v.ph_next is None
self.heap = ph_pairing(v.ph_child)
v.ph_child = None
self.heap = ph_pairing(self.heap.ph_child)
return v
def remove(self, v):
@@ -122,7 +123,6 @@ class Task:
def __init__(self, coro, globals=None):
self.coro = coro # Coroutine of this Task
self.data = None # General data for queue it is waiting on
self.state = True # None, False, True, a callable, or a TaskQueue instance
self.ph_key = 0 # Pairing heap
self.ph_child = None # Paring heap
self.ph_child_last = None # Paring heap
@@ -130,33 +130,30 @@ class Task:
self.ph_rightmost_parent = None # Paring heap
def __iter__(self):
if not self.state:
# Task finished, signal that is has been await'ed on.
self.state = False
elif self.state is True:
# Allocated head of linked list of Tasks waiting on completion of this task.
self.state = TaskQueue()
elif type(self.state) is not TaskQueue:
# Task has state used for another purpose, so can't also wait on it.
raise RuntimeError("can't wait")
if self.coro is self:
# Signal that the completed-task has been await'ed on.
self.waiting = None
elif not hasattr(self, "waiting"):
# Lazily allocated head of linked list of Tasks waiting on completion of this task.
self.waiting = TaskQueue()
return self
def __next__(self):
if not self.state:
if self.coro is self:
# Task finished, raise return value to caller so it can continue.
raise self.data
else:
# Put calling task on waiting queue.
self.state.push(core.cur_task)
self.waiting.push_head(core.cur_task)
# Set calling task's data to this task that it waits on, to double-link it.
core.cur_task.data = self
def done(self):
return not self.state
return self.coro is self
def cancel(self):
# Check if task is already finished.
if not self.state:
if self.coro is self:
return False
# Can't cancel self (not supported yet).
if self is core.cur_task:
@@ -168,10 +165,20 @@ class Task:
if hasattr(self.data, "remove"):
# Not on the main running queue, remove the task from the queue it's on.
self.data.remove(self)
core._task_queue.push(self)
core._task_queue.push_head(self)
elif core.ticks_diff(self.ph_key, core.ticks()) > 0:
# On the main running queue but scheduled in the future, so bring it forward to now.
core._task_queue.remove(self)
core._task_queue.push(self)
core._task_queue.push_head(self)
self.data = core.CancelledError
return True
def throw(self, value):
# This task raised an exception which was uncaught; handle that now.
# Set the data because it was cleared by the main scheduling loop.
self.data = value
if not hasattr(self, "waiting"):
# Nothing await'ed on the task so call the exception handler.
core._exc_context["exception"] = value
core._exc_context["future"] = self
core.Loop.call_exception_handler(core._exc_context)

View File

@@ -1,18 +1,12 @@
import io
import os
import sys
try:
import traceback
except ImportError:
traceback = None
class SkipTest(Exception):
pass
class AssertRaisesContext:
def __init__(self, exc):
self.expected = exc
@@ -20,98 +14,29 @@ class AssertRaisesContext:
return self
def __exit__(self, exc_type, exc_value, tb):
self.exception = exc_value
if exc_type is None:
assert False, "%r not raised" % self.expected
if issubclass(exc_type, self.expected):
# store exception for later retrieval
self.exception = exc_value
return True
return False
# These are used to provide required context to things like subTest
__current_test__ = None
__test_result__ = None
class SubtestContext:
def __init__(self, msg=None, params=None):
self.msg = msg
self.params = params
def __enter__(self):
pass
def __exit__(self, *exc_info):
if exc_info[0] is not None:
# Exception raised
global __test_result__, __current_test__
test_details = __current_test__
if self.msg:
test_details += (f" [{self.msg}]",)
if self.params:
detail = ", ".join(f"{k}={v}" for k, v in self.params.items())
test_details += (f" ({detail})",)
_handle_test_exception(test_details, __test_result__, exc_info, False)
# Suppress the exception as we've captured it above
return True
class NullContext:
def __enter__(self):
pass
def __exit__(self, exc_type, exc_value, traceback):
pass
class TestCase:
def __init__(self):
pass
def addCleanup(self, func, *args, **kwargs):
if not hasattr(self, "_cleanups"):
self._cleanups = []
self._cleanups.append((func, args, kwargs))
def doCleanups(self):
if hasattr(self, "_cleanups"):
while self._cleanups:
func, args, kwargs = self._cleanups.pop()
func(*args, **kwargs)
def subTest(self, msg=None, **params):
return SubtestContext(msg=msg, params=params)
def skipTest(self, reason):
raise SkipTest(reason)
def fail(self, msg=""):
def fail(self, msg=''):
assert False, msg
def assertEqual(self, x, y, msg=""):
def assertEqual(self, x, y, msg=''):
if not msg:
msg = "%r vs (expected) %r" % (x, y)
assert x == y, msg
def assertNotEqual(self, x, y, msg=""):
def assertNotEqual(self, x, y, msg=''):
if not msg:
msg = "%r not expected to be equal %r" % (x, y)
assert x != y, msg
def assertLessEqual(self, x, y, msg=None):
if msg is None:
msg = "%r is expected to be <= %r" % (x, y)
assert x <= y, msg
def assertGreaterEqual(self, x, y, msg=None):
if msg is None:
msg = "%r is expected to be >= %r" % (x, y)
assert x >= y, msg
def assertAlmostEqual(self, x, y, places=None, msg="", delta=None):
def assertAlmostEqual(self, x, y, places=None, msg='', delta=None):
if x == y:
return
if delta is not None and places is not None:
@@ -121,18 +46,18 @@ class TestCase:
if abs(x - y) <= delta:
return
if not msg:
msg = "%r != %r within %r delta" % (x, y, delta)
msg = '%r != %r within %r delta' % (x, y, delta)
else:
if places is None:
places = 7
if round(abs(y - x), places) == 0:
if round(abs(y-x), places) == 0:
return
if not msg:
msg = "%r != %r within %r places" % (x, y, places)
msg = '%r != %r within %r places' % (x, y, places)
assert False, msg
def assertNotAlmostEqual(self, x, y, places=None, msg="", delta=None):
def assertNotAlmostEqual(self, x, y, places=None, msg='', delta=None):
if delta is not None and places is not None:
raise TypeError("specify delta or places not both")
@@ -140,53 +65,53 @@ class TestCase:
if not (x == y) and abs(x - y) > delta:
return
if not msg:
msg = "%r == %r within %r delta" % (x, y, delta)
msg = '%r == %r within %r delta' % (x, y, delta)
else:
if places is None:
places = 7
if not (x == y) and round(abs(y - x), places) != 0:
if not (x == y) and round(abs(y-x), places) != 0:
return
if not msg:
msg = "%r == %r within %r places" % (x, y, places)
msg = '%r == %r within %r places' % (x, y, places)
assert False, msg
def assertIs(self, x, y, msg=""):
def assertIs(self, x, y, msg=''):
if not msg:
msg = "%r is not %r" % (x, y)
assert x is y, msg
def assertIsNot(self, x, y, msg=""):
def assertIsNot(self, x, y, msg=''):
if not msg:
msg = "%r is %r" % (x, y)
assert x is not y, msg
def assertIsNone(self, x, msg=""):
def assertIsNone(self, x, msg=''):
if not msg:
msg = "%r is not None" % x
assert x is None, msg
def assertIsNotNone(self, x, msg=""):
def assertIsNotNone(self, x, msg=''):
if not msg:
msg = "%r is None" % x
assert x is not None, msg
def assertTrue(self, x, msg=""):
def assertTrue(self, x, msg=''):
if not msg:
msg = "Expected %r to be True" % x
assert x, msg
def assertFalse(self, x, msg=""):
def assertFalse(self, x, msg=''):
if not msg:
msg = "Expected %r to be False" % x
assert not x, msg
def assertIn(self, x, y, msg=""):
def assertIn(self, x, y, msg=''):
if not msg:
msg = "Expected %r to be in %r" % (x, y)
assert x in y, msg
def assertIsInstance(self, x, y, msg=""):
def assertIsInstance(self, x, y, msg=''):
assert isinstance(x, y), msg
def assertRaises(self, exc, func=None, *args, **kwargs):
@@ -195,15 +120,12 @@ class TestCase:
try:
func(*args, **kwargs)
assert False, "%r not raised" % exc
except Exception as e:
if isinstance(e, exc):
return
raise
assert False, "%r not raised" % exc
def assertWarns(self, warn):
return NullContext()
def skip(msg):
@@ -211,252 +133,92 @@ def skip(msg):
# We just replace original fun with _inner
def _inner(self):
raise SkipTest(msg)
return _inner
return _decor
def skipIf(cond, msg):
if not cond:
return lambda x: x
return skip(msg)
def skipUnless(cond, msg):
if cond:
return lambda x: x
return skip(msg)
def expectedFailure(test):
def test_exp_fail(*args, **kwargs):
try:
test(*args, **kwargs)
except:
pass
else:
assert False, "unexpected success"
return test_exp_fail
class TestSuite:
def __init__(self, name=""):
self._tests = []
self.name = name
def __init__(self):
self.tests = []
def addTest(self, cls):
self._tests.append(cls)
def run(self, result):
for c in self._tests:
_run_suite(c, result, self.name)
return result
def _load_module(self, mod):
for tn in dir(mod):
c = getattr(mod, tn)
if isinstance(c, object) and isinstance(c, type) and issubclass(c, TestCase):
self.addTest(c)
elif tn.startswith("test") and callable(c):
self.addTest(c)
self.tests.append(cls)
class TestRunner:
def run(self, suite: TestSuite):
def run(self, suite):
res = TestResult()
suite.run(res)
for c in suite.tests:
run_class(c, res)
res.printErrors()
print("----------------------------------------------------------------------")
print("Ran %d tests\n" % res.testsRun)
if res.failuresNum > 0 or res.errorsNum > 0:
print("FAILED (failures=%d, errors=%d)" % (res.failuresNum, res.errorsNum))
else:
msg = "OK"
if res.skippedNum > 0:
msg += " (skipped=%d)" % res.skippedNum
msg += " (%d skipped)" % res.skippedNum
print(msg)
return res
TextTestRunner = TestRunner
class TestResult:
def __init__(self):
self.errorsNum = 0
self.failuresNum = 0
self.skippedNum = 0
self.testsRun = 0
self.errors = []
self.failures = []
self.skipped = []
self._newFailures = 0
def wasSuccessful(self):
return self.errorsNum == 0 and self.failuresNum == 0
def printErrors(self):
print()
self.printErrorList(self.errors)
self.printErrorList(self.failures)
def printErrorList(self, lst):
sep = "----------------------------------------------------------------------"
for c, e in lst:
detail = " ".join((str(i) for i in c))
print("======================================================================")
print(f"FAIL: {detail}")
print(sep)
print(e)
def __repr__(self):
# Format is compatible with CPython.
return "<unittest.result.TestResult run=%d errors=%d failures=%d>" % (
self.testsRun,
self.errorsNum,
self.failuresNum,
)
def __add__(self, other):
self.errorsNum += other.errorsNum
self.failuresNum += other.failuresNum
self.skippedNum += other.skippedNum
self.testsRun += other.testsRun
self.errors.extend(other.errors)
self.failures.extend(other.failures)
self.skipped.extend(other.skipped)
return self
def _capture_exc(exc, exc_traceback):
buf = io.StringIO()
if hasattr(sys, "print_exception"):
sys.print_exception(exc, buf)
elif traceback is not None:
traceback.print_exception(None, exc, exc_traceback, file=buf)
return buf.getvalue()
def _handle_test_exception(
current_test: tuple, test_result: TestResult, exc_info: tuple, verbose=True
):
exc = exc_info[1]
traceback = exc_info[2]
ex_str = _capture_exc(exc, traceback)
if isinstance(exc, AssertionError):
test_result.failuresNum += 1
test_result.failures.append((current_test, ex_str))
if verbose:
print(" FAIL")
else:
test_result.errorsNum += 1
test_result.errors.append((current_test, ex_str))
if verbose:
print(" ERROR")
test_result._newFailures += 1
def _run_suite(c, test_result: TestResult, suite_name=""):
if isinstance(c, TestSuite):
c.run(test_result)
return
if isinstance(c, type):
o = c()
else:
o = c
set_up_class = getattr(o, "setUpClass", lambda: None)
tear_down_class = getattr(o, "tearDownClass", lambda: None)
# TODO: Uncompliant
def run_class(c, test_result):
o = c()
set_up = getattr(o, "setUp", lambda: None)
tear_down = getattr(o, "tearDown", lambda: None)
exceptions = []
try:
suite_name += "." + c.__qualname__
except AttributeError:
pass
def run_one(test_function):
global __test_result__, __current_test__
print("%s (%s) ..." % (name, suite_name), end="")
set_up()
__test_result__ = test_result
test_container = f"({suite_name})"
__current_test__ = (name, test_container)
try:
test_result._newFailures = 0
test_result.testsRun += 1
test_function()
# No exception occurred, test passed
if test_result._newFailures:
print(" FAIL")
else:
print(" ok")
except SkipTest as e:
reason = e.args[0]
print(" skipped:", reason)
test_result.skippedNum += 1
test_result.skipped.append((name, c, reason))
except Exception as ex:
_handle_test_exception(
current_test=(name, c), test_result=test_result, exc_info=(type(ex), ex, None)
)
# Uncomment to investigate failure in detail
# raise
finally:
__test_result__ = None
__current_test__ = None
tear_down()
for name in dir(o):
if name.startswith("test"):
print("%s (%s) ..." % (name, c.__qualname__), end="")
m = getattr(o, name)
set_up()
try:
o.doCleanups()
except AttributeError:
pass
set_up_class()
try:
if hasattr(o, "runTest"):
name = str(o)
run_one(o.runTest)
return
for name in dir(o):
if name.startswith("test"):
m = getattr(o, name)
if not callable(m):
continue
run_one(m)
if callable(o):
name = o.__name__
run_one(o)
finally:
tear_down_class()
return exceptions
test_result.testsRun += 1
m()
print(" ok")
except SkipTest as e:
print(" skipped:", e.args[0])
test_result.skippedNum += 1
except:
print(" FAIL")
test_result.failuresNum += 1
# Uncomment to investigate failure in detail
#raise
continue
finally:
tear_down()
# This supports either:
#
# >>> import mytest
# >>> unitttest.main(mytest)
#
# >>> unittest.main("mytest")
#
# Or, a script that ends with:
# if __name__ == "__main__":
# unittest.main()
# e.g. run via `mpremote run mytest.py`
def main(module="__main__", testRunner=None):
if testRunner is None:
testRunner = TestRunner()
elif isinstance(testRunner, type):
testRunner = testRunner()
def main(module="__main__"):
def test_cases(m):
for tn in dir(m):
c = getattr(m, tn)
if isinstance(c, object) and isinstance(c, type) and issubclass(c, TestCase):
yield c
if isinstance(module, str):
module = __import__(module)
suite = TestSuite(module.__name__)
suite._load_module(module)
return testRunner.run(suite)
m = __import__(module)
suite = TestSuite()
for c in test_cases(m):
suite.addTest(c)
runner = TestRunner()
result = runner.run(suite)
# Terminate with non zero return code in case of failures
sys.exit(result.failuresNum > 0)

View File

@@ -1,57 +1,6 @@
[project]
name = "microdot"
version = "1.3.5.dev0"
authors = [
{ name = "Miguel Grinberg", email = "miguel.grinberg@gmail.com" },
]
description = "The impossibly small web framework for MicroPython"
classifiers = [
"Environment :: Web Environment",
"Intended Audience :: Developers",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: Implementation :: MicroPython",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
]
[project.readme]
file = "README.md"
content-type = "text/markdown"
[project.urls]
Homepage = "https://github.com/miguelgrinberg/microdot"
"Bug Tracker" = "https://github.com/miguelgrinberg/microdot/issues"
[project.optional-dependencies]
docs = [
"sphinx",
]
[tool.setuptools]
zip-safe = false
include-package-data = true
py-modules = [
"microdot",
"microdot_asyncio",
"microdot_utemplate",
"microdot_jinja",
"microdot_session",
"microdot_cors",
"microdot_websocket",
"microdot_websocket_alt",
"microdot_asyncio_websocket",
"microdot_test_client",
"microdot_asyncio_test_client",
"microdot_wsgi",
"microdot_asgi",
"microdot_asgi_websocket",
]
[tool.setuptools.package-dir]
"" = "src"
[build-system]
requires = [
"setuptools>=61.2",
"setuptools>=42",
"wheel"
]
build-backend = "setuptools.build_meta"

View File

@@ -6,5 +6,4 @@ sys.path.insert(3, 'libs/micropython')
import unittest
if not unittest.main('tests').wasSuccessful():
sys.exit(1)
unittest.main('tests')

40
setup.cfg Normal file
View File

@@ -0,0 +1,40 @@
[metadata]
name = microdot
version = 1.1.2.dev0
author = Miguel Grinberg
author_email = miguel.grinberg@gmail.com
description = The impossibly small web framework for MicroPython
long_description = file: README.md
long_description_content_type = text/markdown
url = https://github.com/miguelgrinberg/microdot
project_urls =
Bug Tracker = https://github.com/miguelgrinberg/microdot/issues
classifiers =
Environment :: Web Environment
Intended Audience :: Developers
Programming Language :: Python :: 3
Programming Language :: Python :: Implementation :: MicroPython
License :: OSI Approved :: MIT License
Operating System :: OS Independent
[options]
zip_safe = False
include_package_data = True
package_dir =
= src
py_modules =
microdot
microdot_asyncio
microdot_utemplate
microdot_jinja
microdot_session
microdot_auth
microdot_login
microdot_websocket
microdot_websocket_alt
microdot_asyncio_websocket
microdot_test_client
microdot_asyncio_test_client
microdot_wsgi
microdot_asgi
microdot_asgi_websocket

3
setup.py Executable file
View File

@@ -0,0 +1,3 @@
import setuptools
setuptools.setup()

View File

@@ -43,13 +43,11 @@ try:
except ImportError:
import re
socket_timeout_error = OSError
try:
import usocket as socket
except ImportError:
try:
import socket
socket_timeout_error = socket.timeout
except ImportError: # pragma: no cover
socket = None
@@ -94,9 +92,8 @@ def urldecode_bytes(s):
def urlencode(s):
return s.replace('+', '%2B').replace(' ', '+').replace(
'%', '%25').replace('?', '%3F').replace('#', '%23').replace(
'&', '%26').replace('=', '%3D')
return s.replace(' ', '+').replace('%', '%25').replace('?', '%3F').replace(
'#', '%23').replace('&', '%26').replace('+', '%2B')
class NoCaseDict(dict):
@@ -146,43 +143,6 @@ class NoCaseDict(dict):
kl = key.lower()
return super().get(self.keymap.get(kl, kl), default)
def update(self, other_dict):
for key, value in other_dict.items():
self[key] = value
def mro(cls): # pragma: no cover
"""Return the method resolution order of a class.
This is a helper function that returns the method resolution order of a
class. It is used by Microdot to find the best error handler to invoke for
the raised exception.
In CPython, this function returns the ``__mro__`` attribute of the class.
In MicroPython, this function implements a recursive depth-first scanning
of the class hierarchy.
"""
if hasattr(cls, 'mro'):
return cls.__mro__
def _mro(cls):
m = [cls]
for base in cls.__bases__:
m += _mro(base)
return m
mro_list = _mro(cls)
# If a class appears multiple times (due to multiple inheritance) remove
# all but the last occurence. This matches the method resolution order
# of MicroPython, but not CPython.
mro_pruned = []
for i in range(len(mro_list)):
base = mro_list.pop(0)
if base not in mro_list:
mro_pruned.append(base)
return mro_pruned
class MultiDict(dict):
"""A subclass of dictionary that can hold multiple values for the same
@@ -306,12 +266,6 @@ class Request():
#: Request.max_readline = 16 * 1024 # 16KB lines allowed
max_readline = 2 * 1024
#: Specify a suggested read timeout to use when reading the request. Set to
#: 0 to disable the use of a timeout. This timeout should be considered a
#: suggestion only, as some platforms may not support it. The default is
#: 1 second.
socket_read_timeout = 1
class G:
pass
@@ -404,15 +358,13 @@ class Request():
data = MultiDict()
if len(urlencoded) > 0:
if isinstance(urlencoded, str):
for kv in [pair.split('=', 1)
for pair in urlencoded.split('&') if pair]:
data[urldecode_str(kv[0])] = urldecode_str(kv[1]) \
if len(kv) > 1 else ''
for k, v in [pair.split('=', 1)
for pair in urlencoded.split('&')]:
data[urldecode_str(k)] = urldecode_str(v)
elif isinstance(urlencoded, bytes): # pragma: no branch
for kv in [pair.split(b'=', 1)
for pair in urlencoded.split(b'&') if pair]:
data[urldecode_bytes(kv[0])] = urldecode_bytes(kv[1]) \
if len(kv) > 1 else b''
for k, v in [pair.split(b'=', 1)
for pair in urlencoded.split(b'&')]:
data[urldecode_bytes(k)] = urldecode_bytes(v)
return data
@property
@@ -486,9 +438,6 @@ class Request():
return response
return 'Hello, World!'
Note that the function is not called if the request handler raises an
exception and an error response is returned instead.
"""
self.after_request_handlers.append(f)
return f
@@ -532,10 +481,6 @@ class Response():
#: ``Content-Type`` header.
default_content_type = 'text/plain'
#: The default cache control max age used by :meth:`send_file`. A value
#: of ``None`` means that no ``Cache-Control`` header is added.
default_send_file_max_age = None
#: Special response used to signal that a response does not need to be
#: written to the client. Used to exit WebSocket connections cleanly.
already_handled = None
@@ -555,7 +500,6 @@ class Response():
else:
# this applies to bytes, file-like objects or generators
self.body = body
self.is_head = False
def set_cookie(self, cookie, value, path=None, domain=None, expires=None,
max_age=None, secure=False, http_only=False):
@@ -620,20 +564,19 @@ class Response():
stream.write(b'\r\n')
# body
if not self.is_head:
can_flush = hasattr(stream, 'flush')
try:
for body in self.body_iter():
if isinstance(body, str): # pragma: no cover
body = body.encode()
stream.write(body)
if can_flush: # pragma: no cover
stream.flush()
except OSError as exc: # pragma: no cover
if exc.errno in MUTED_SOCKET_ERRORS:
pass
else:
raise
can_flush = hasattr(stream, 'flush')
try:
for body in self.body_iter():
if isinstance(body, str): # pragma: no cover
body = body.encode()
stream.write(body)
if can_flush: # pragma: no cover
stream.flush()
except OSError as exc: # pragma: no cover
if exc.errno in MUTED_SOCKET_ERRORS:
pass
else:
raise
def body_iter(self):
if self.body:
@@ -664,9 +607,7 @@ class Response():
return cls(status_code=status_code, headers={'Location': location})
@classmethod
def send_file(cls, filename, status_code=200, content_type=None,
stream=None, max_age=None, compressed=False,
file_extension=''):
def send_file(cls, filename, status_code=200, content_type=None):
"""Send file contents in a response.
:param filename: The filename of the file.
@@ -674,25 +615,7 @@ class Response():
default is 302.
:param content_type: The ``Content-Type`` header to use in the
response. If omitted, it is generated
automatically from the file extension of the
``filename`` parameter.
:param stream: A file-like object to read the file contents from. If
a stream is given, the ``filename`` parameter is only
used when generating the ``Content-Type`` header.
:param max_age: The ``Cache-Control`` header's ``max-age`` value in
seconds. If omitted, the value of the
:attr:`Response.default_send_file_max_age` attribute is
used.
:param compressed: Whether the file is compressed. If ``True``, the
``Content-Encoding`` header is set to ``gzip``. A
string with the header value can also be passed.
Note that when using this option the file must have
been compressed beforehand. This option only sets
the header.
:param file_extension: A file extension to append to the ``filename``
parameter when opening the file, including the
dot. The extension given here is not considered
when generating the ``Content-Type`` header.
automatically from the file extension.
Security note: The filename is assumed to be trusted. Never pass
filenames provided by the user without validating and sanitizing them
@@ -704,19 +627,9 @@ class Response():
content_type = Response.types_map[ext]
else:
content_type = 'application/octet-stream'
headers = {'Content-Type': content_type}
if max_age is None:
max_age = cls.default_send_file_max_age
if max_age is not None:
headers['Cache-Control'] = 'max-age={}'.format(max_age)
if compressed:
headers['Content-Encoding'] = compressed \
if isinstance(compressed, str) else 'gzip'
f = stream or open(filename + file_extension, 'rb')
return cls(body=f, status_code=status_code, headers=headers)
f = open(filename, 'rb')
return cls(body=f, status_code=status_code,
headers={'Content-Type': content_type})
class URLPattern():
@@ -738,7 +651,7 @@ class URLPattern():
if type_ == 'string':
pattern = '[^/]+'
elif type_ == 'int':
pattern = '-?\\d+'
pattern = '\\d+'
elif type_ == 'path':
pattern = '.+'
elif type_.startswith('re:'):
@@ -799,10 +712,8 @@ class Microdot():
self.url_map = []
self.before_request_handlers = []
self.after_request_handlers = []
self.after_error_request_handlers = []
self.error_handlers = {}
self.shutdown_requested = False
self.options_handler = self.default_options_handler
self.debug = False
self.server = None
@@ -838,8 +749,7 @@ class Microdot():
"""
def decorated(f):
self.url_map.append(
([m.upper() for m in (methods or ['GET'])],
URLPattern(url_pattern), f))
(methods or ['GET'], URLPattern(url_pattern), f))
return f
return decorated
@@ -963,24 +873,6 @@ class Microdot():
self.after_request_handlers.append(f)
return f
def after_error_request(self, f):
"""Decorator to register a function to run after an error response is
generated. The decorated function must take two arguments, the request
and response objects. The return value of the function must be an
updated response object. The handler is invoked for error responses
generated by Microdot, as well as those returned by application-defined
error handlers.
Example::
@app.after_error_request
def func(request, response):
# ...
return response
"""
self.after_error_request_handlers.append(f)
return f
def errorhandler(self, status_code_or_exception_class):
"""Decorator to register a function as an error handler. Error handler
functions for numeric HTTP status codes must accept a single argument,
@@ -1021,8 +913,6 @@ class Microdot():
self.before_request_handlers.append(handler)
for handler in subapp.after_request_handlers:
self.after_request_handlers.append(handler)
for handler in subapp.after_error_request_handlers:
self.after_error_request_handlers.append(handler)
for status_code, handler in subapp.error_handlers.items():
self.error_handlers[status_code] = handler
@@ -1074,7 +964,7 @@ class Microdot():
app = Microdot()
@app.route('/')
def index(request):
def index():
return 'Hello, world!'
app.run(debug=True)
@@ -1125,36 +1015,18 @@ class Microdot():
self.shutdown_requested = True
def find_route(self, req):
method = req.method.upper()
if method == 'OPTIONS' and self.options_handler:
return self.options_handler(req)
if method == 'HEAD':
method = 'GET'
f = 404
for route_methods, route_pattern, route_handler in self.url_map:
req.url_args = route_pattern.match(req.path)
if req.url_args is not None:
if method in route_methods:
if req.method in route_methods:
f = route_handler
break
else:
f = 405
return f
def default_options_handler(self, req):
allow = []
for route_methods, route_pattern, route_handler in self.url_map:
if route_pattern.match(req.path) is not None:
allow.extend(route_methods)
if 'GET' in allow:
allow.append('HEAD')
allow.append('OPTIONS')
return {'Allow': ', '.join(allow)}
def handle_request(self, sock, addr):
if Request.socket_read_timeout and \
hasattr(sock, 'settimeout'): # pragma: no cover
sock.settimeout(Request.socket_read_timeout)
if not hasattr(sock, 'readline'): # pragma: no cover
stream = sock.makefile("rwb")
else:
@@ -1165,9 +1037,6 @@ class Microdot():
try:
req = Request.create(self, stream, addr, sock)
res = self.dispatch_request(req)
except socket_timeout_error as exc: # pragma: no cover
if exc.errno and exc.errno != errno.ETIMEDOUT:
print_exception(exc) # not a timeout
except Exception as exc: # pragma: no cover
print_exception(exc)
try:
@@ -1191,7 +1060,6 @@ class Microdot():
status_code=res.status_code))
def dispatch_request(self, req):
after_request_handled = False
if req:
if req.content_length > req.max_content_length:
if 413 in self.error_handlers:
@@ -1224,9 +1092,6 @@ class Microdot():
res = handler(req, res) or res
for handler in req.after_request_handlers:
res = handler(req, res) or res
after_request_handled = True
elif isinstance(f, dict):
res = Response(headers=f)
elif f in self.error_handlers:
res = self.error_handlers[f](req)
else:
@@ -1238,18 +1103,10 @@ class Microdot():
res = exc.reason, exc.status_code
except Exception as exc:
print_exception(exc)
exc_class = None
res = None
if exc.__class__ in self.error_handlers:
exc_class = exc.__class__
else:
for c in mro(exc.__class__)[1:]:
if c in self.error_handlers:
exc_class = c
break
if exc_class:
try:
res = self.error_handlers[exc_class](req, exc)
res = self.error_handlers[exc.__class__](req, exc)
except Exception as exc2: # pragma: no cover
print_exception(exc2)
if res is None:
@@ -1267,10 +1124,6 @@ class Microdot():
res = Response(*res)
elif not isinstance(res, Response):
res = Response(res)
if not after_request_handled:
for handler in self.after_error_request_handlers:
res = handler(req, res) or res
res.is_head = (req and req.method == 'HEAD')
return res

View File

@@ -59,9 +59,8 @@ class Microdot(BaseMicrodot):
headers = NoCaseDict()
content_length = 0
for key, value in scope.get('headers', []):
key = key.decode().title()
headers[key] = value.decode()
if key == 'Content-Length':
headers[key] = value
if key.lower() == 'content-length':
content_length = int(value)
if content_length and content_length <= Request.max_body_length:
@@ -94,10 +93,10 @@ class Microdot(BaseMicrodot):
header_list = []
for name, value in res.headers.items():
if not isinstance(value, list):
header_list.append((name.lower().encode(), value.encode()))
header_list.append((name, value))
else:
for v in value:
header_list.append((name.lower().encode(), v.encode()))
header_list.append((name, v))
if scope['type'] != 'http': # pragma: no cover
return
@@ -120,18 +119,17 @@ class Microdot(BaseMicrodot):
asyncio.ensure_future(cancel_monitor())
body_iter = res.body_iter().__aiter__()
res_body = b''
try:
res_body = await body_iter.__anext__()
body = await body_iter.__anext__()
while not cancelled: # pragma: no branch
next_body = await body_iter.__anext__()
await send({'type': 'http.response.body',
'body': res_body,
'body': body,
'more_body': True})
res_body = next_body
body = next_body
except StopAsyncIteration:
await send({'type': 'http.response.body',
'body': res_body,
'body': body,
'more_body': False})
async def __call__(self, scope, receive, send):

View File

@@ -17,7 +17,6 @@ except ImportError:
import io
from microdot import Microdot as BaseMicrodot
from microdot import mro
from microdot import NoCaseDict
from microdot import Request as BaseRequest
from microdot import Response as BaseResponse
@@ -151,11 +150,10 @@ class Response(BaseResponse):
await stream.awrite(b'\r\n')
# body
if not self.is_head:
async for body in self.body_iter():
if isinstance(body, str): # pragma: no cover
body = body.encode()
await stream.awrite(body)
async for body in self.body_iter():
if isinstance(body, str): # pragma: no cover
body = body.encode()
await stream.awrite(body)
except OSError as exc: # pragma: no cover
if exc.errno in MUTED_SOCKET_ERRORS or \
exc.args[0] == 'Connection lost':
@@ -241,7 +239,7 @@ class Microdot(BaseMicrodot):
app = Microdot()
@app.route('/')
async def index(request):
async def index():
return 'Hello, world!'
async def main():
@@ -280,11 +278,6 @@ class Microdot(BaseMicrodot):
while True:
try:
if hasattr(self.server, 'serve_forever'): # pragma: no cover
try:
await self.server.serve_forever()
except asyncio.CancelledError:
pass
await self.server.wait_closed()
break
except AttributeError: # pragma: no cover
@@ -318,7 +311,7 @@ class Microdot(BaseMicrodot):
app = Microdot()
@app.route('/')
async def index(request):
async def index():
return 'Hello, world!'
app.run(debug=True)
@@ -353,7 +346,6 @@ class Microdot(BaseMicrodot):
status_code=res.status_code))
async def dispatch_request(self, req):
after_request_handled = False
if req:
if req.content_length > req.max_content_length:
if 413 in self.error_handlers:
@@ -388,11 +380,7 @@ class Microdot(BaseMicrodot):
res = await self._invoke_handler(
handler, req, res) or res
for handler in req.after_request_handlers:
res = await self._invoke_handler(
handler, req, res) or res
after_request_handled = True
elif isinstance(f, dict):
res = Response(headers=f)
res = await handler(req, res) or res
elif f in self.error_handlers:
res = await self._invoke_handler(
self.error_handlers[f], req)
@@ -405,19 +393,11 @@ class Microdot(BaseMicrodot):
res = exc.reason, exc.status_code
except Exception as exc:
print_exception(exc)
exc_class = None
res = None
if exc.__class__ in self.error_handlers:
exc_class = exc.__class__
else:
for c in mro(exc.__class__)[1:]:
if c in self.error_handlers:
exc_class = c
break
if exc_class:
try:
res = await self._invoke_handler(
self.error_handlers[exc_class], req, exc)
self.error_handlers[exc.__class__], req, exc)
except Exception as exc2: # pragma: no cover
print_exception(exc2)
if res is None:
@@ -435,11 +415,6 @@ class Microdot(BaseMicrodot):
res = Response(*res)
elif not isinstance(res, Response):
res = Response(res)
if not after_request_handled:
for handler in self.after_error_request_handlers:
res = await self._invoke_handler(
handler, req, res) or res
res.is_head = (req and req.method == 'HEAD')
return res
async def _invoke_handler(self, f_or_coro, *args, **kwargs):

View File

@@ -17,7 +17,7 @@ class WebSocket(BaseWebSocket):
opcode, payload = await self._read_frame()
send_opcode, data = self._process_websocket_frame(opcode, payload)
if send_opcode: # pragma: no cover
await self.send(data, send_opcode)
await self.send(send_opcode, data)
elif data: # pragma: no branch
return data

65
src/microdot_auth.py Normal file
View File

@@ -0,0 +1,65 @@
from microdot import abort
class BaseAuth:
def __init__(self, header='Authorization', scheme=None):
self.auth_callback = None
self.error_callback = self.auth_failed
self.header = header
self.scheme = scheme.lower()
def callback(self, f):
"""Decorator to configure the authentication callback.
Microdot calls the authentication callback to allow the application to
check user credentials.
"""
self.auth_callback = f
def errorhandler(self, f):
"""Decorator to configure the error callback.
Microdot calls the error callback to allow the application to generate
a custom error response. The default error response is to call
``abort(401)``.
"""
self.error_callback = f
def auth_failed(self):
abort(401)
def __call__(self, func):
def wrapper(request, *args, **kwargs):
auth = request.headers.get(self.header)
if not auth:
return self.error_callback()
if self.header == 'Authorization':
if ' ' not in auth:
return self.error_callback()
scheme, auth = auth.split(' ', 1)
if scheme.lower() != self.scheme:
return self.error_callback()
if not self.auth_callback(request, *self._get_auth_args(auth)):
return self.error_callback()
return func(request, *args, **kwargs)
return wrapper
class BasicAuth(BaseAuth):
def __init__(self):
super().__init__(scheme='Basic')
def _get_auth_args(self, auth):
import binascii
username, password = binascii.a2b_base64(auth).decode('utf-8').split(
':', 1)
return (username, password)
class TokenAuth(BaseAuth):
def __init__(self, header='Authorization', scheme='Bearer'):
super().__init__(header=header, scheme=scheme)
def _get_auth_args(self, token):
return (token,)

View File

@@ -1,110 +0,0 @@
class CORS:
"""Add CORS headers to HTTP responses.
:param app: The application to add CORS headers to.
:param allowed_origins: A list of origins that are allowed to make
cross-site requests. If set to '*', all origins are
allowed.
:param allow_credentials: If set to True, the
``Access-Control-Allow-Credentials`` header will
be set to ``true`` to indicate to the browser
that it can expose cookies and authentication
headers.
:param allowed_methods: A list of methods that are allowed to be used when
making cross-site requests. If not set, all methods
are allowed.
:param expose_headers: A list of headers that the browser is allowed to
exposed.
:param allowed_headers: A list of headers that are allowed to be used when
making cross-site requests. If not set, all headers
are allowed.
:param max_age: The maximum amount of time in seconds that the browser
should cache the results of a preflight request.
:param handle_cors: If set to False, CORS headers will not be added to
responses. This can be useful if you want to add CORS
headers manually.
"""
def __init__(self, app=None, allowed_origins=None, allow_credentials=False,
allowed_methods=None, expose_headers=None,
allowed_headers=None, max_age=None, handle_cors=True):
self.allowed_origins = allowed_origins
self.allow_credentials = allow_credentials
self.allowed_methods = allowed_methods
self.expose_headers = expose_headers
self.allowed_headers = None if allowed_headers is None \
else [h.lower() for h in allowed_headers]
self.max_age = max_age
if app is not None:
self.initialize(app, handle_cors=handle_cors)
def initialize(self, app, handle_cors=True):
"""Initialize the CORS object for the given application.
:param app: The application to add CORS headers to.
:param handle_cors: If set to False, CORS headers will not be added to
responses. This can be useful if you want to add
CORS headers manually.
"""
self.default_options_handler = app.options_handler
if handle_cors:
app.options_handler = self.options_handler
app.after_request(self.after_request)
app.after_error_request(self.after_request)
def options_handler(self, request):
headers = self.default_options_handler(request)
headers.update(self.get_cors_headers(request))
return headers
def get_cors_headers(self, request):
"""Return a dictionary of CORS headers to add to a given request.
:param request: The request to add CORS headers to.
"""
cors_headers = {}
origin = request.headers.get('Origin')
if self.allowed_origins == '*':
cors_headers['Access-Control-Allow-Origin'] = origin or '*'
if origin:
cors_headers['Vary'] = 'Origin'
elif origin in (self.allowed_origins or []):
cors_headers['Access-Control-Allow-Origin'] = origin
cors_headers['Vary'] = 'Origin'
if self.allow_credentials and \
'Access-Control-Allow-Origin' in cors_headers:
cors_headers['Access-Control-Allow-Credentials'] = 'true'
if self.expose_headers:
cors_headers['Access-Control-Expose-Headers'] = \
', '.join(self.expose_headers)
if request.method == 'OPTIONS':
# handle preflight request
if self.max_age:
cors_headers['Access-Control-Max-Age'] = str(self.max_age)
method = request.headers.get('Access-Control-Request-Method')
if method:
method = method.upper()
if self.allowed_methods is None or \
method in self.allowed_methods:
cors_headers['Access-Control-Allow-Methods'] = method
headers = request.headers.get('Access-Control-Request-Headers')
if headers:
if self.allowed_headers is None:
cors_headers['Access-Control-Allow-Headers'] = headers
else:
headers = [h.strip() for h in headers.split(',')]
headers = [h for h in headers
if h.lower() in self.allowed_headers]
cors_headers['Access-Control-Allow-Headers'] = \
', '.join(headers)
return cors_headers
def after_request(self, request, response):
saved_vary = response.headers.get('Vary')
response.headers.update(self.get_cors_headers(request))
if saved_vary and saved_vary != response.headers.get('Vary'):
response.headers['Vary'] = (
saved_vary + ', ' + response.headers['Vary'])

46
src/microdot_login.py Normal file
View File

@@ -0,0 +1,46 @@
from microdot import redirect, urlencode
from microdot_session import get_session, update_session
class LoginAuth:
def __init__(self, login_url='/login'):
super().__init__()
self.login_url = login_url
self.user_callback = self._accept_user
def callback(self, f):
self.user_callback = f
def login_user(self, request, user_id):
session = get_session(request)
session['user_id'] = user_id
update_session(request, session)
return session
def logout_user(self, request):
session = get_session(request)
session.pop('user_id', None)
update_session(request, session)
return session
def redirect_to_next(self, request, default_url='/'):
next_url = request.args.get('next', default_url)
if not next_url.startswith('/'):
next_url = default_url
return redirect(next_url)
def __call__(self, func):
def wrapper(request, *args, **kwargs):
session = get_session(request)
if 'user_id' not in session:
return redirect(self.login_url + '?next=' + urlencode(
request.url))
if not self.user_callback(request, session['user_id']):
return redirect(self.login_url + '?next=' + urlencode(
request.url))
return func(request, *args, **kwargs)
return wrapper
def _accept_user(self, request, user_id):
return True

View File

@@ -58,7 +58,6 @@ class TestResponse:
test_res._initialize_body(res)
test_res._process_text_body()
test_res._process_json_body()
test_res.is_head = res.is_head
return test_res

View File

@@ -28,7 +28,7 @@ class WebSocket:
opcode, payload = self._read_frame()
send_opcode, data = self._process_websocket_frame(opcode, payload)
if send_opcode: # pragma: no cover
self.send(data, send_opcode)
self.send(send_opcode, data)
elif data: # pragma: no branch
return data

Binary file not shown.

View File

@@ -1,158 +0,0 @@
import unittest
from microdot import Microdot
from microdot_test_client import TestClient
from microdot_cors import CORS
class TestCORS(unittest.TestCase):
def test_origin(self):
app = Microdot()
cors = CORS(allowed_origins=['https://example.com'],
allow_credentials=True)
cors.initialize(app)
@app.get('/')
def index(req):
return 'foo'
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 200)
self.assertFalse('Access-Control-Allow-Origin' in res.headers)
self.assertFalse('Access-Control-Allow-Credentials' in res.headers)
self.assertFalse('Vary' in res.headers)
res = client.get('/', headers={'Origin': 'https://example.com'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
self.assertEqual(res.headers['Access-Control-Allow-Credentials'],
'true')
self.assertEqual(res.headers['Vary'], 'Origin')
cors.allow_credentials = False
res = client.get('/foo', headers={'Origin': 'https://example.com'})
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
self.assertFalse('Access-Control-Allow-Credentials' in res.headers)
self.assertEqual(res.headers['Vary'], 'Origin')
res = client.get('/', headers={'Origin': 'https://bad.com'})
self.assertEqual(res.status_code, 200)
self.assertFalse('Access-Control-Allow-Origin' in res.headers)
self.assertFalse('Access-Control-Allow-Credentials' in res.headers)
self.assertFalse('Vary' in res.headers)
def test_all_origins(self):
app = Microdot()
CORS(app, allowed_origins='*', expose_headers=['X-Test', 'X-Test2'])
@app.get('/')
def index(req):
return 'foo'
@app.get('/foo')
def foo(req):
return 'foo', {'Vary': 'X-Foo, X-Bar'}
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'], '*')
self.assertFalse('Vary' in res.headers)
self.assertEqual(res.headers['Access-Control-Expose-Headers'],
'X-Test, X-Test2')
res = client.get('/', headers={'Origin': 'https://example.com'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
self.assertEqual(res.headers['Vary'], 'Origin')
self.assertEqual(res.headers['Access-Control-Expose-Headers'],
'X-Test, X-Test2')
res = client.get('/bad', headers={'Origin': 'https://example.com'})
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
self.assertEqual(res.headers['Vary'], 'Origin')
self.assertEqual(res.headers['Access-Control-Expose-Headers'],
'X-Test, X-Test2')
res = client.get('/foo', headers={'Origin': 'https://example.com'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Vary'], 'X-Foo, X-Bar, Origin')
def test_cors_preflight(self):
app = Microdot()
CORS(app, allowed_origins='*')
@app.route('/', methods=['GET', 'POST'])
def index(req):
return 'foo'
client = TestClient(app)
res = client.request('OPTIONS', '/', headers={
'Origin': 'https://example.com',
'Access-Control-Request-Method': 'POST',
'Access-Control-Request-Headers': 'X-Test, X-Test2'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
self.assertFalse('Access-Control-Max-Age' in res.headers)
self.assertEqual(res.headers['Access-Control-Allow-Methods'], 'POST')
self.assertEqual(res.headers['Access-Control-Allow-Headers'],
'X-Test, X-Test2')
res = client.request('OPTIONS', '/', headers={
'Origin': 'https://example.com'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
self.assertFalse('Access-Control-Max-Age' in res.headers)
self.assertFalse('Access-Control-Allow-Methods' in res.headers)
self.assertFalse('Access-Control-Allow-Headers' in res.headers)
def test_cors_preflight_with_options(self):
app = Microdot()
CORS(app, allowed_origins='*', max_age=3600, allowed_methods=['POST'],
allowed_headers=['X-Test'])
@app.route('/', methods=['GET', 'POST'])
def index(req):
return 'foo'
client = TestClient(app)
res = client.request('OPTIONS', '/', headers={
'Origin': 'https://example.com',
'Access-Control-Request-Method': 'POST',
'Access-Control-Request-Headers': 'X-Test, X-Test2'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
self.assertEqual(res.headers['Access-Control-Max-Age'], '3600')
self.assertEqual(res.headers['Access-Control-Allow-Methods'], 'POST')
self.assertEqual(res.headers['Access-Control-Allow-Headers'], 'X-Test')
res = client.request('OPTIONS', '/', headers={
'Origin': 'https://example.com',
'Access-Control-Request-Method': 'GET'})
self.assertEqual(res.status_code, 200)
self.assertFalse('Access-Control-Allow-Methods' in res.headers)
self.assertFalse('Access-Control-Allow-Headers' in res.headers)
def test_cors_disabled(self):
app = Microdot()
CORS(app, allowed_origins='*', handle_cors=False)
@app.get('/')
def index(req):
return 'foo'
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 200)
self.assertFalse('Access-Control-Allow-Origin' in res.headers)
self.assertFalse('Vary' in res.headers)

View File

@@ -63,52 +63,6 @@ class TestMicrodot(unittest.TestCase):
self.assertEqual(res.headers['Content-Length'], '3')
self.assertEqual(res.text, 'bar')
def test_head_request(self):
self._mock()
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
mock_socket.clear_requests()
fd = mock_socket.add_request('HEAD', '/foo')
self._add_shutdown(app)
app.run()
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n'))
self._unmock()
def test_options_request(self):
app = Microdot()
@app.route('/', methods=['GET', 'DELETE'])
def index(req):
return 'foo'
@app.post('/')
def index_post(req):
return 'bar'
@app.route('/foo', methods=['POST', 'PUT'])
def foo(req):
return 'baz'
client = TestClient(app)
res = client.request('OPTIONS', '/')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'],
'GET, DELETE, POST, HEAD, OPTIONS')
res = client.request('OPTIONS', '/foo')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'], 'POST, PUT, OPTIONS')
def test_empty_request(self):
self._mock()
@@ -325,39 +279,6 @@ class TestMicrodot(unittest.TestCase):
self.assertEqual(res.headers['Content-Length'], '3')
self.assertEqual(res.text, 'baz')
def test_after_error_request(self):
app = Microdot()
@app.after_error_request
def after_error_request_one(req, res):
res.headers['X-One'] = '1'
@app.after_error_request
def after_error_request_two(req, res):
res.set_cookie('foo', 'bar')
return res
@app.route('/foo')
def foo(req):
return 'foo'
client = TestClient(app)
res = client.get('/foo')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertFalse('X-One' in res.headers)
self.assertFalse('Set-Cookie' in res.headers)
res = client.get('/bar')
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.headers['Set-Cookie'], ['foo=bar'])
self.assertEqual(res.headers['X-One'], '1')
self.assertEqual(client.cookies['foo'], 'bar')
def test_400(self):
self._mock()
@@ -544,90 +465,6 @@ class TestMicrodot(unittest.TestCase):
'text/plain; charset=UTF-8')
self.assertEqual(res.text, '501')
def test_exception_handler_parent(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(LookupError)
def handle_lookup_error(req, exc):
return exc.__class__.__name__, 501
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'IndexError')
def test_exception_handler_redundant_parent(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(LookupError)
def handle_lookup_error(req, exc):
return 'LookupError', 501
@app.errorhandler(IndexError)
def handle_index_error(req, exc):
return 'IndexError', 501
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'IndexError')
def test_exception_handler_multiple_parents(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(Exception)
def handle_generic_exception(req, exc):
return 'Exception', 501
@app.errorhandler(LookupError)
def handle_lookup_error(req, exc):
return 'LookupError', 501
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'LookupError')
def test_exception_handler_no_viable_parents(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(RuntimeError)
def handle_runtime_error(req, exc):
return 'RuntimeError', 501
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 500)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'Internal server error')
def test_abort(self):
app = Microdot()
@@ -740,11 +577,7 @@ class TestMicrodot(unittest.TestCase):
@subapp.after_request
def after(req, res):
res.body += b':after'
@subapp.after_error_request
def after_error(req, res):
res.body += b':errorafter'
return res.body + b':after'
@subapp.errorhandler(404)
def not_found(req):
@@ -763,7 +596,7 @@ class TestMicrodot(unittest.TestCase):
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, '404:errorafter')
self.assertEqual(res.text, '404')
res = client.get('/sub/app')
self.assertEqual(res.status_code, 200)

View File

@@ -53,9 +53,9 @@ class TestMicrodotASGI(unittest.TestCase):
'type': 'http',
'path': '/foo/bar',
'query_string': b'baz=1',
'headers': [(b'Authorization', b'Bearer 123'),
(b'Cookie', b'session=xyz'),
(b'Content-Length', b'4')],
'headers': [('Authorization', 'Bearer 123'),
('Cookie', 'session=xyz'),
('Content-Length', 4)],
'client': ['1.2.3.4', 1234],
'method': 'POST',
'http_version': '1.1',
@@ -83,10 +83,10 @@ class TestMicrodotASGI(unittest.TestCase):
if packet['type'] == 'http.response.start':
self.assertEqual(packet['status'], 200)
expected_headers = [
(b'content-length', b'8'),
(b'content-type', b'text/plain; charset=UTF-8'),
(b'set-cookie', b'foo=foo'),
(b'set-cookie', b'bar=bar; HttpOnly')
('Content-Length', '8'),
('Content-Type', 'text/plain; charset=UTF-8'),
('Set-Cookie', 'foo=foo'),
('Set-Cookie', 'bar=bar; HttpOnly')
]
self.assertEqual(len(packet['headers']), len(expected_headers))
for header in expected_headers:
@@ -114,9 +114,9 @@ class TestMicrodotASGI(unittest.TestCase):
scope = {
'type': 'http',
'path': '/foo/bar',
'headers': [(b'Authorization', b'Bearer 123'),
(b'Cookie', b'session=xyz'),
(b'Content-Length', b'4')],
'headers': [('Authorization', 'Bearer 123'),
('Cookie', 'session=xyz'),
('Content-Length', 4)],
'client': ['1.2.3.4', 1234],
'method': 'POST',
'http_version': '1.1',

View File

@@ -101,48 +101,6 @@ class TestMicrodotAsync(unittest.TestCase):
self.assertEqual(res.body, b'bar-async')
self.assertEqual(res.json, None)
def test_head_request(self):
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
mock_socket.clear_requests()
fd = mock_socket.add_request('HEAD', '/foo')
self._add_shutdown(app)
app.run()
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n'))
def test_options_request(self):
app = Microdot()
@app.route('/', methods=['GET', 'DELETE'])
async def index(req):
return 'foo'
@app.post('/')
async def index_post(req):
return 'bar'
@app.route('/foo', methods=['POST', 'PUT'])
async def foo(req):
return 'baz'
client = TestClient(app)
res = self._run(client.request('OPTIONS', '/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'],
'GET, DELETE, POST, HEAD, OPTIONS')
res = self._run(client.request('OPTIONS', '/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'], 'POST, PUT, OPTIONS')
def test_empty_request(self):
app = Microdot()
@@ -356,39 +314,6 @@ class TestMicrodotAsync(unittest.TestCase):
self.assertEqual(res.headers['Content-Length'], '3')
self.assertEqual(res.text, 'baz')
def test_after_error_request(self):
app = Microdot()
@app.after_error_request
def after_error_request_one(req, res):
res.headers['X-One'] = '1'
@app.after_error_request
def after_error_request_two(req, res):
res.set_cookie('foo', 'bar')
return res
@app.route('/foo')
def foo(req):
return 'foo'
client = TestClient(app)
res = self._run(client.get('/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertFalse('X-One' in res.headers)
self.assertFalse('Set-Cookie' in res.headers)
res = self._run(client.get('/bar'))
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.headers['Set-Cookie'], ['foo=bar'])
self.assertEqual(res.headers['X-One'], '1')
self.assertEqual(client.cookies['foo'], 'bar')
def test_400(self):
self._mock()
@@ -575,90 +500,6 @@ class TestMicrodotAsync(unittest.TestCase):
'text/plain; charset=UTF-8')
self.assertEqual(res.text, '501')
def test_exception_handler_parent(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(LookupError)
async def handle_lookup_error(req, exc):
return exc.__class__.__name__, 501
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'IndexError')
def test_exception_handler_redundant_parent(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(LookupError)
async def handle_lookup_error(req, exc):
return 'LookupError', 501
@app.errorhandler(IndexError)
async def handle_index_error(req, exc):
return 'IndexError', 501
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'IndexError')
def test_exception_handler_multiple_parents(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(Exception)
async def handle_generic_exception(req, exc):
return 'Exception', 501
@app.errorhandler(LookupError)
async def handle_lookup_error(req, exc):
return 'LookupError', 501
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'LookupError')
def test_exception_handler_no_viable_parents(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(RuntimeError)
async def handle_runtime_error(req, exc):
return 'RuntimeError', 501
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 500)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'Internal server error')
def test_abort(self):
app = Microdot()

113
tests/test_microdot_auth.py Normal file
View File

@@ -0,0 +1,113 @@
import binascii
import unittest
from microdot import Microdot
from microdot_auth import BasicAuth, TokenAuth
from microdot_test_client import TestClient
class TestAuth(unittest.TestCase):
def test_basic_auth(self):
app = Microdot()
basic_auth = BasicAuth()
@basic_auth.callback
def authenticate(request, username, password):
if username == 'foo' and password == 'bar':
request.g.user = {'username': username}
return True
@app.route('/')
@basic_auth
def index(request):
return request.g.user['username']
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 401)
res = client.get('/', headers={
'Authorization': 'Basic ' + binascii.b2a_base64(
b'foo:bar').decode()})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'foo')
res = client.get('/', headers={
'Authorization': 'Basic ' + binascii.b2a_base64(
b'foo:baz').decode()})
self.assertEqual(res.status_code, 401)
def test_token_auth(self):
app = Microdot()
token_auth = TokenAuth()
@token_auth.callback
def authenticate(request, token):
if token == 'foo':
request.g.user = 'user'
return True
@app.route('/')
@token_auth
def index(request):
return request.g.user
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 401)
res = client.get('/', headers={'Authorization': 'Basic foo'})
self.assertEqual(res.status_code, 401)
res = client.get('/', headers={'Authorization': 'foo'})
self.assertEqual(res.status_code, 401)
res = client.get('/', headers={'Authorization': 'Bearer foo'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'user')
def test_token_auth_custom_header(self):
app = Microdot()
token_auth = TokenAuth(header='X-Auth-Token')
@token_auth.callback
def authenticate(request, token):
if token == 'foo':
request.g.user = 'user'
return True
@app.route('/')
@token_auth
def index(request):
return request.g.user
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 401)
res = client.get('/', headers={'Authorization': 'Basic foo'})
self.assertEqual(res.status_code, 401)
res = client.get('/', headers={'Authorization': 'foo'})
self.assertEqual(res.status_code, 401)
res = client.get('/', headers={'Authorization': 'Bearer foo'})
self.assertEqual(res.status_code, 401)
res = client.get('/', headers={'X-Token-Auth': 'Bearer foo'})
self.assertEqual(res.status_code, 401)
res = client.get('/', headers={'X-Auth-Token': 'foo'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'user')
res = client.get('/', headers={'x-auth-token': 'foo'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'user')
@token_auth.errorhandler
def error_handler():
return {'status_code': 403}, 403
res = client.get('/')
self.assertEqual(res.status_code, 403)
self.assertEqual(res.json, {'status_code': 403})

View File

@@ -0,0 +1,134 @@
import unittest
from microdot import Microdot
from microdot_login import LoginAuth
from microdot_session import set_session_secret_key, with_session
from microdot_test_client import TestClient
set_session_secret_key('top-secret!')
class TestLogin(unittest.TestCase):
def test_login_auth(self):
app = Microdot()
login_auth = LoginAuth()
@app.get('/')
@login_auth
def index(request):
return 'ok'
@app.post('/login')
def login(request):
login_auth.login_user(request, 'user')
return login_auth.redirect_to_next(request)
@app.post('/logout')
def logout(request):
login_auth.logout_user(request)
return 'ok'
client = TestClient(app)
res = client.get('/?foo=bar')
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/login?next=/%3Ffoo%3Dbar')
res = client.post('/login?next=/%3Ffoo=bar')
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/?foo=bar')
res = client.get('/')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'ok')
res = client.post('/logout')
self.assertEqual(res.status_code, 200)
res = client.get('/')
self.assertEqual(res.status_code, 302)
def test_login_auth_with_session(self):
app = Microdot()
login_auth = LoginAuth(login_url='/foo')
@app.get('/')
@login_auth
@with_session
def index(request, session):
return session['user_id']
@app.post('/foo')
def login(request):
login_auth.login_user(request, 'user')
return login_auth.redirect_to_next(request)
client = TestClient(app)
res = client.get('/')
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/foo?next=/')
res = client.post('/foo')
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/')
res = client.get('/')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'user')
def test_login_auth_user_callback(self):
app = Microdot()
login_auth = LoginAuth()
@login_auth.callback
def check_user(request, user_id):
request.g.user_id = user_id
return user_id == 'user'
@app.get('/')
@login_auth
def index(request):
return request.g.user_id
@app.post('/good-login')
def good_login(request):
login_auth.login_user(request, 'user')
return login_auth.redirect_to_next(request)
@app.post('/bad-login')
def bad_login(request):
login_auth.login_user(request, 'foo')
return login_auth.redirect_to_next(request)
client = TestClient(app)
res = client.post('/good-login')
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/')
res = client.get('/')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'user')
res = client.post('/bad-login')
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/')
res = client.get('/')
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/login?next=/')
def test_login_auth_bad_redirect(self):
app = Microdot()
login_auth = LoginAuth()
@app.get('/')
@login_auth
def index(request):
return 'ok'
@app.post('/login')
def login(request):
login_auth.login_user(request, 'user')
return login_auth.redirect_to_next(request)
client = TestClient(app)
res = client.post('/login?next=http://example.com')
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/')

View File

@@ -58,11 +58,3 @@ class TestMultiDict(unittest.TestCase):
del d['oNE']
self.assertEqual(list(d.items()), [('two', 5)])
self.assertEqual(list(d.values()), [5])
d.update({'oNe': 1, 'two': 2, 'three': 3})
self.assertEqual(d['one'], 1)
self.assertEqual(d['ONE'], 1)
self.assertEqual(d['two'], 2)
self.assertEqual(d['TWO'], 2)
self.assertEqual(d['three'], 3)
self.assertEqual(d['THREE'], 3)

View File

@@ -39,17 +39,9 @@ class TestRequest(unittest.TestCase):
self.assertEqual(req.body, b'aaa')
def test_args(self):
fd = get_request_fd('GET', '/?foo=bar&abc=def&foo&x=%2f%%')
fd = get_request_fd('GET', '/?foo=bar&abc=def&x=%2f%%')
req = Request.create('app', fd, 'addr')
self.assertEqual(req.query_string, 'foo=bar&abc=def&foo&x=%2f%%')
md = MultiDict({'foo': 'bar', 'abc': 'def', 'x': '/%%'})
md['foo'] = ''
self.assertEqual(req.args, md)
def test_badly_formatted_args(self):
fd = get_request_fd('GET', '/?&foo=bar&abc=def&&&x=%2f%%')
req = Request.create('app', fd, 'addr')
self.assertEqual(req.query_string, '&foo=bar&abc=def&&&x=%2f%%')
self.assertEqual(req.query_string, 'foo=bar&abc=def&x=%2f%%')
self.assertEqual(req.args, MultiDict(
{'foo': 'bar', 'abc': 'def', 'x': '/%%'}))

View File

@@ -49,12 +49,11 @@ class TestRequestAsync(unittest.TestCase):
self.assertEqual(req.body, b'aaa')
def test_args(self):
fd = get_async_request_fd('GET', '/?foo=bar&abc=def&foo&x=%2f%%')
fd = get_async_request_fd('GET', '/?foo=bar&abc=def&x=%2f%%')
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.query_string, 'foo=bar&abc=def&foo&x=%2f%%')
md = MultiDict({'foo': 'bar', 'abc': 'def', 'x': '/%%'})
md['foo'] = ''
self.assertEqual(req.args, md)
self.assertEqual(req.query_string, 'foo=bar&abc=def&x=%2f%%')
self.assertEqual(req.args, MultiDict(
{'foo': 'bar', 'abc': 'def', 'x': '/%%'}))
def test_json(self):
fd = get_async_request_fd('GET', '/foo', headers={

View File

@@ -235,39 +235,6 @@ class TestResponse(unittest.TestCase):
b'HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\nfoo\n')
Response.send_file_buffer_size = original_buffer_size
def test_send_file_max_age(self):
res = Response.send_file('tests/files/test.txt', max_age=123)
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Cache-Control'], 'max-age=123')
Response.default_send_file_max_age = 456
res = Response.send_file('tests/files/test.txt')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Cache-Control'], 'max-age=456')
res = Response.send_file('tests/files/test.txt', max_age=123)
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Cache-Control'], 'max-age=123')
Response.default_send_file_max_age = None
def test_send_file_compressed(self):
res = Response.send_file('tests/files/test.txt', compressed=True)
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'], 'text/plain')
self.assertEqual(res.headers['Content-Encoding'], 'gzip')
res = Response.send_file('tests/files/test.txt', compressed='foo')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'], 'text/plain')
self.assertEqual(res.headers['Content-Encoding'], 'foo')
res = Response.send_file('tests/files/test', compressed=True,
file_extension='.gz')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'application/octet-stream')
self.assertEqual(res.headers['Content-Encoding'], 'gzip')
def test_default_content_type(self):
original_content_type = Response.default_content_type
res = Response('foo')

View File

@@ -1,24 +1,22 @@
try:
import uasyncio as asyncio
except ImportError:
import asyncio
import unittest
from microdot import Microdot
from microdot_asyncio import Microdot as MicrodotAsync
from microdot_session import set_session_secret_key, get_session, \
update_session, delete_session, with_session
from microdot_test_client import TestClient
from microdot_asyncio_test_client import TestClient as TestClientAsync
set_session_secret_key('top-secret!')
class TestSession(unittest.TestCase):
def test_session(self):
app = Microdot()
client = TestClient(app)
def setUp(self):
self.app = Microdot()
self.client = TestClient(self.app)
@app.get('/')
def tearDown(self):
pass
def test_session(self):
@self.app.get('/')
def index(req):
session = get_session(req)
session2 = get_session(req)
@@ -26,106 +24,52 @@ class TestSession(unittest.TestCase):
self.assertEqual(session['foo'], 'bar')
return str(session.get('name'))
@app.get('/with')
@self.app.get('/with')
@with_session
def session_context_manager(req, session):
return str(session.get('name'))
@app.post('/set')
@self.app.post('/set')
def set_session(req):
update_session(req, {'name': 'joe'})
return 'OK'
@app.post('/del')
@self.app.post('/del')
def del_session(req):
delete_session(req)
return 'OK'
res = client.get('/')
res = self.client.get('/')
self.assertEqual(res.text, 'None')
res = client.get('/with')
res = self.client.get('/with')
self.assertEqual(res.text, 'None')
res = client.post('/set')
res = self.client.post('/set')
self.assertEqual(res.text, 'OK')
res = client.get('/')
res = self.client.get('/')
self.assertEqual(res.text, 'joe')
res = client.get('/with')
res = self.client.get('/with')
self.assertEqual(res.text, 'joe')
res = client.post('/del')
res = self.client.post('/del')
self.assertEqual(res.text, 'OK')
res = client.get('/')
res = self.client.get('/')
self.assertEqual(res.text, 'None')
res = client.get('/with')
self.assertEqual(res.text, 'None')
def _run(self, coro):
loop = asyncio.get_event_loop()
return loop.run_until_complete(coro)
def test_session_async(self):
app = MicrodotAsync()
client = TestClientAsync(app)
@app.get('/')
async def index(req):
session = get_session(req)
session2 = get_session(req)
session2['foo'] = 'bar'
self.assertEqual(session['foo'], 'bar')
return str(session.get('name'))
@app.get('/with')
@with_session
async def session_context_manager(req, session):
return str(session.get('name'))
@app.post('/set')
async def set_session(req):
update_session(req, {'name': 'joe'})
return 'OK'
@app.post('/del')
async def del_session(req):
delete_session(req)
return 'OK'
res = self._run(client.get('/'))
self.assertEqual(res.text, 'None')
res = self._run(client.get('/with'))
self.assertEqual(res.text, 'None')
res = self._run(client.post('/set'))
self.assertEqual(res.text, 'OK')
res = self._run(client.get('/'))
self.assertEqual(res.text, 'joe')
res = self._run(client.get('/with'))
self.assertEqual(res.text, 'joe')
res = self._run(client.post('/del'))
self.assertEqual(res.text, 'OK')
res = self._run(client.get('/'))
self.assertEqual(res.text, 'None')
res = self._run(client.get('/with'))
res = self.client.get('/with')
self.assertEqual(res.text, 'None')
def test_session_no_secret_key(self):
set_session_secret_key(None)
app = Microdot()
client = TestClient(app)
@app.get('/')
@self.app.get('/')
def index(req):
self.assertRaises(ValueError, get_session, req)
self.assertRaises(ValueError, update_session, req, {})
return ''
res = client.get('/')
res = self.client.get('/')
self.assertEqual(res.status_code, 200)
set_session_secret_key('top-secret!')

View File

@@ -56,12 +56,10 @@ class TestURLPattern(unittest.TestCase):
p = URLPattern('/users/<int:id>/<int:id2>/')
self.assertEqual(p.match('/users/123/456/'), {'id': 123, 'id2': 456})
self.assertEqual(p.match('/users/123/-456/'), {'id': 123, 'id2': -456})
self.assertIsNone(p.match('/users/'))
self.assertIsNone(p.match('/users/123/-456'))
self.assertIsNone(p.match('/users/123/456'))
self.assertIsNone(p.match('/users/123/abc/'))
self.assertIsNone(p.match('/users/123/-456/abc'))
self.assertIsNone(p.match('/users/--123/456/'))
self.assertIsNone(p.match('/users/123/456/abc'))
def test_path_argument(self):
p = URLPattern('/users/<path:path>')

View File

@@ -1,11 +0,0 @@
import unittest
from microdot import urlencode, urldecode_str, urldecode_bytes
class TestURLEncode(unittest.TestCase):
def test_urlencode(self):
self.assertEqual(urlencode('?foo=bar&x'), '%3Ffoo%3Dbar%26x')
def test_urldecode(self):
self.assertEqual(urldecode_str('%3Ffoo%3Dbar%26x'), '?foo=bar&x')
self.assertEqual(urldecode_bytes(b'%3Ffoo%3Dbar%26x'), '?foo=bar&x')

View File

@@ -1,4 +1,4 @@
FROM ubuntu:22.04
FROM ubuntu:20.04
ARG DEBIAN_FRONTEND=noninteractive

View File

@@ -1,10 +1,6 @@
#!/bin/bash
# this script updates the micropython binary in the /bin directory that is
# used to run unit tests under GitHub Actions builds
DOCKER=${DOCKER:-docker}
$DOCKER build -t micropython .
$DOCKER create -it --name dummy-micropython micropython
$DOCKER cp dummy-micropython:/usr/local/bin/micropython ../bin/micropython
$DOCKER rm dummy-micropython
docker build -t micropython .
docker create -it --name dummy-micropython micropython
docker cp dummy-micropython:/usr/local/bin/micropython ../bin/micropython
docker rm dummy-micropython

10
tox.ini
View File

@@ -1,21 +1,21 @@
[tox]
envlist=flake8,py37,py38,py39,py310,py311,upy,benchmark
envlist=flake8,py36,py37,py38,py39,py310,upy,benchmark
skipsdist=True
skip_missing_interpreters=True
[gh-actions]
python =
3.6: py36
3.7: py37
3.8: py38
3.9: py39
3.10: py310
3.11: py311
pypy3: pypy3
[testenv]
commands=
pip install -e .
pytest -p no:logging --cov=src --cov-config=.coveragerc --cov-branch --cov-report=term-missing --cov-report=xml
pytest -p no:logging --cov=src --cov-config=.coveragerc --cov-branch --cov-report=term-missing
deps=
pytest
pytest-cov
@@ -31,11 +31,11 @@ commands=
flake8 --ignore=W503 --exclude src/utemplate,tests/libs src tests examples
[testenv:upy]
allowlist_externals=sh
whitelist_externals=sh
commands=sh -c "bin/micropython run_tests.py"
[testenv:upy-mac]
allowlist_externals=micropython
whitelist_externals=micropython
commands=micropython run_tests.py
deps=