Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions docs/api/bo4e.com.rst
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,14 @@ bo4e.com.angebotsposition module
:undoc-members:
:show-inheritance:

bo4e.com.angebotsteil module
----------------------------

.. automodule:: bo4e.com.angebotsteil
:members:
:undoc-members:
:show-inheritance:

bo4e.com.aufabschlag module
---------------------------

Expand Down
94 changes: 94 additions & 0 deletions src/bo4e/com/angebotsteil.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
"""
Contains Angebotsteil class
and corresponding marshmallow schema for de-/serialization
"""

from typing import List, Optional

import attr
from marshmallow import fields, post_load

from bo4e.bo.marktlokation import Marktlokation, MarktlokationSchema
from bo4e.com.angebotsposition import Angebotsposition, AngebotspositionSchema
from bo4e.com.betrag import Betrag, BetragSchema
from bo4e.com.com import COM, COMSchema
from bo4e.com.menge import Menge, MengeSchema
from bo4e.com.zeitraum import Zeitraum, ZeitraumSchema
from bo4e.validators import check_list_length_at_least_one


# pylint: disable=too-few-public-methods
@attr.s(auto_attribs=True, kw_only=True)
class Angebotsteil(COM):
"""
Mit dieser Komponente wird ein Teil einer Angebotsvariante abgebildet.
Hier werden alle Angebotspositionen aggregiert.
Angebotsteile werden im einfachsten Fall für eine Marktlokation oder Lieferstellenadresse erzeugt.
Hier werden die Mengen und Gesamtkosten aller Angebotspositionen zusammengefasst.
Eine Variante besteht mindestens aus einem Angebotsteil.
"""

# required attributes
#: Einzelne Positionen, die zu diesem Angebotsteil gehören
positionen: List[Angebotsposition] = attr.ib(
validator=[
attr.validators.deep_iterable(
member_validator=attr.validators.instance_of(Angebotsposition),
iterable_validator=attr.validators.instance_of(list),
),
check_list_length_at_least_one,
]
)

# optional attributes
#: Identifizierung eines Subkapitels einer Anfrage, beispielsweise das Los einer Ausschreibung
anfrage_subreferenz: Optional[str] = attr.ib(
default=None, validator=attr.validators.optional(attr.validators.instance_of(str))
)
lieferstellenangebotsteil: Optional[List[Marktlokation]] = attr.ib(
default=None,
validator=attr.validators.optional(
attr.validators.deep_iterable(
member_validator=attr.validators.instance_of(Marktlokation),
iterable_validator=attr.validators.instance_of(list),
)
),
)
"""
Marktlokationen, für die dieses Angebotsteil gilt, falls vorhanden.
Durch die Marktlokation ist auch die Lieferadresse festgelegt
"""
#: Summe der Verbräuche aller in diesem Angebotsteil eingeschlossenen Lieferstellen
gesamtmengeangebotsteil: Optional[Menge] = attr.ib(
default=None, validator=attr.validators.optional(attr.validators.instance_of(Menge))
)
#: Summe der Jahresenergiekosten aller in diesem Angebotsteil enthaltenen Lieferstellen
gesamtkostenangebotsteil: Optional[Betrag] = attr.ib(
default=None, validator=attr.validators.optional(attr.validators.instance_of(Betrag))
)
#: Hier kann der Belieferungszeitraum angegeben werden, für den dieser Angebotsteil gilt
lieferzeitraum: Optional[Zeitraum] = attr.ib(
default=None, validator=attr.validators.optional(attr.validators.instance_of(Zeitraum))
)


class AngebotsteilSchema(COMSchema):
"""
Schema for de-/serialization of Angebotsteil.
"""

# required attributes
positionen = fields.List(fields.Nested(AngebotspositionSchema))

# optional attributes
anfrage_subreferenz = fields.Str(load_default=None)
lieferstellenangebotsteil = fields.List(fields.Nested(MarktlokationSchema), load_default=None)
gesamtmengeangebotsteil = fields.Nested(MengeSchema, load_default=None)
gesamtkostenangebotsteil = fields.Nested(BetragSchema, load_default=None)
lieferzeitraum = fields.Nested(ZeitraumSchema, load_default=None)

# pylint: disable=no-self-use, unused-argument
@post_load
def deserialize(self, data, **kwargs) -> Angebotsteil:
"""Deserialize JSON to Angebotsteil object"""
return Angebotsteil(**data)
187 changes: 187 additions & 0 deletions tests/test_angebotsteil.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,187 @@
from datetime import datetime, timezone
from decimal import Decimal

import pytest # type:ignore[import]

from bo4e.bo.marktlokation import Marktlokation
from bo4e.com.adresse import Adresse
from bo4e.com.angebotsposition import Angebotsposition
from bo4e.com.angebotsteil import Angebotsteil, AngebotsteilSchema
from bo4e.com.betrag import Betrag
from bo4e.com.menge import Menge
from bo4e.com.preis import Preis
from bo4e.com.zeitraum import Zeitraum
from bo4e.enum.bilanzierungsmethode import Bilanzierungsmethode
from bo4e.enum.energierichtung import Energierichtung
from bo4e.enum.mengeneinheit import Mengeneinheit
from bo4e.enum.netzebene import Netzebene
from bo4e.enum.sparte import Sparte
from bo4e.enum.waehrungscode import Waehrungscode
from bo4e.enum.waehrungseinheit import Waehrungseinheit
from tests.serialization_helper import assert_serialization_roundtrip # type:ignore[import]


class TestAngebotsteil:
@pytest.mark.parametrize(
"angebotsteil, expected_json_dict",
[
pytest.param(
Angebotsteil(
positionen=[
Angebotsposition(
positionsbezeichnung="testtring",
positionsmenge=Menge(wert=Decimal(4000), einheit=Mengeneinheit.KWH),
positionspreis=Preis(
wert=Decimal(0.2456), einheit=Waehrungseinheit.EUR, bezugswert=Mengeneinheit.KWH
),
positionskosten=Betrag(
waehrung=Waehrungscode.EUR,
wert=Decimal(98240),
),
)
],
anfrage_subreferenz="teststring",
lieferstellenangebotsteil=[
Marktlokation(
marktlokations_id="51238696781",
sparte=Sparte.GAS,
lokationsadresse=Adresse(
postleitzahl="82031",
ort="Grünwald",
hausnummer="27A",
strasse="Nördliche Münchner Straße",
),
energierichtung=Energierichtung.EINSP,
bilanzierungsmethode=Bilanzierungsmethode.PAUSCHAL,
unterbrechbar=True,
netzebene=Netzebene.NSP,
)
],
gesamtmengeangebotsteil=Menge(wert=Decimal(4000), einheit=Mengeneinheit.KWH),
gesamtkostenangebotsteil=Betrag(
waehrung=Waehrungscode.EUR,
wert=Decimal(98240),
),
lieferzeitraum=Zeitraum(
startdatum=datetime(2020, 1, 1, tzinfo=timezone.utc),
enddatum=datetime(2020, 4, 1, tzinfo=timezone.utc),
),
),
{
"positionen": [
{
"positionsbezeichnung": "testtring",
"positionsmenge": {"wert": "4000", "einheit": "KWH"},
"positionskosten": {"waehrung": "EUR", "wert": "98240"},
"positionspreis": {
"bezugswert": "KWH",
"status": None,
"wert": "0.2456000000000000127453603226967970840632915496826171875",
"einheit": "EUR",
},
},
],
"lieferstellenangebotsteil": [
{
"marktlokationsId": "51238696781",
"sparte": "GAS",
"lokationsadresse": {
"postleitzahl": "82031",
"ort": "Grünwald",
"hausnummer": "27A",
"strasse": "Nördliche Münchner Straße",
"adresszusatz": None,
"postfach": None,
"coErgaenzung": None,
"landescode": "DE",
},
"energierichtung": "EINSP",
"bilanzierungsmethode": "PAUSCHAL",
"unterbrechbar": True,
"netzebene": "NSP",
"netzgebietsnr": None,
"versionstruktur": "2",
"katasterinformation": None,
"bilanzierungsgebiet": None,
"grundversorgercodenr": None,
"endkunde": None,
"geoadresse": None,
"verbrauchsart": None,
"netzbetreibercodenr": None,
"gebietstyp": None,
"gasqualitaet": None,
"zugehoerigeMesslokation": None,
"externeReferenzen": [],
"boTyp": "MARKTLOKATION",
}
],
"gesamtmengeangebotsteil": {"wert": "4000", "einheit": "KWH"},
"gesamtkostenangebotsteil": {"waehrung": "EUR", "wert": "98240"},
"anfrageSubreferenz": "teststring",
"lieferzeitraum": {
"startdatum": "2020-01-01T00:00:00+00:00",
"endzeitpunkt": None,
"einheit": None,
"enddatum": "2020-04-01T00:00:00+00:00",
"startzeitpunkt": None,
"dauer": None,
},
},
id="maximal attributes",
),
pytest.param(
Angebotsteil(
positionen=[
Angebotsposition(
positionsbezeichnung="teststring",
positionsmenge=Menge(wert=Decimal(4000), einheit=Mengeneinheit.KWH),
positionspreis=Preis(
wert=Decimal(0.2456), einheit=Waehrungseinheit.EUR, bezugswert=Mengeneinheit.KWH
),
positionskosten=Betrag(
waehrung=Waehrungscode.EUR,
wert=Decimal(98240),
),
)
],
),
{
"positionen": [
{
"positionsbezeichnung": "teststring",
"positionsmenge": {"wert": "4000", "einheit": "KWH"},
"positionskosten": {"waehrung": "EUR", "wert": "98240"},
"positionspreis": {
"bezugswert": "KWH",
"status": None,
"wert": "0.2456000000000000127453603226967970840632915496826171875",
"einheit": "EUR",
},
},
],
"anfrageSubreferenz": None,
"lieferstellenangebotsteil": None,
"gesamtmengeangebotsteil": None,
"gesamtkostenangebotsteil": None,
"lieferzeitraum": None,
},
id="minimal attributes",
),
],
)
def test_serialization_roundtrip(self, angebotsteil, expected_json_dict):
"""
Test de-/serialisation of Angebotsteil with minimal attributes.
"""
assert_serialization_roundtrip(angebotsteil, AngebotsteilSchema(), expected_json_dict)

def test_angebotsteil_positionen_required(self):
with pytest.raises(ValueError) as excinfo:
_ = Angebotsteil(positionen=[])

assert "List positionen must not be empty." in str(excinfo.value)

def test_missing_required_attribute(self):
with pytest.raises(TypeError) as excinfo:
_ = Angebotsteil()
assert "missing 1 required" in str(excinfo.value)