serializers.py

import tarfile
from typing import Dict, Any

from django.core.validators import MinValueValidator
from rest_framework import serializers

from chaincode.models import Chaincode
from chaincode.service import ChaincodeAction, create_chaincode, get_chaincode, install_chaincode, \
    approve_chaincode, commit_chaincode, send_chaincode_request, metadata_exists, get_chaincode_status, \
    get_chaincode_commit_readiness, ChaincodeTransactionError
from channel.models import Channel
from channel.serializers import ChannelID
from common.serializers import ListResponseSerializer
from user.serializers import UserID


class ChaincodeID(serializers.ModelSerializer):
    class Meta:
        model = Chaincode
        fields = ("id",)
        extra_kwargs = {
            # Temporarily make "id" writable only for validation purposes
            "id": {"read_only": False}
        }

    def validate(self, attrs: Dict[str, Any]) -> Dict[str, Any]:
        chaincode = get_chaincode(attrs["id"])
        if chaincode is None:
            raise serializers.ValidationError("Chaincode with id {} does not exist.".format(attrs["id"]))
        self.instance = chaincode
        return attrs

    def update(self, instance: Chaincode, validated_data: Dict[str, Any]) -> Chaincode:
        return instance


class ChaincodeResponse(ChaincodeID):
    channel = ChannelID()
    creator = UserID()
    status = serializers.SerializerMethodField()
    approvals = serializers.SerializerMethodField()

    class Meta:
        model = Chaincode
        fields = (
            "id",
            "name",
            "version",
            "sequence",
            "init_required",
            "signature_policy",
            "package_id",
            "label",
            "creator",
            "channel",
            "language",
            "created_at",
            "description",
            "status",
            "approvals"
        )

    def get_status(self, chaincode) -> str:
        organization = self.context.get("organization")
        return get_chaincode_status(organization, chaincode) if organization else chaincode.get("status")

    def get_approvals(self, chaincode) -> str:
        organization = self.context.get("organization")
        return get_chaincode_commit_readiness(organization, chaincode) if organization else chaincode.get("approvals")


class ChaincodeList(ListResponseSerializer):
    data = ChaincodeResponse(many=True, help_text="Chaincode data")


class ChaincodeCreateBody(serializers.ModelSerializer):
    class Meta:
        model = Chaincode
        fields = (
            "name",
            "version",
            "sequence",
            "init_required",
            "signature_policy",
            "package",
            "channel",
            "description",
        )
        extra_kwargs = {
            "sequence": {
                "validators": [MinValueValidator(1)]
            },
            "init_required": {"required": False},
            "signature_policy": {"required": False},
            "description": {"required": False},
        }

    @staticmethod
    def validate_package(value):
        if not value.name.endswith(".tar.gz"):
            raise serializers.ValidationError("Chaincode Package must be a '.tar.gz' file.")

        if value.content_type != "application/gzip":
            raise serializers.ValidationError(
                "Chaincode Package is not a 'application/gzip' file but {} instead."
                .format(value.content_type)
            )

        try:
            if not metadata_exists(value):
                raise serializers.ValidationError("Metadata not found.")
        except tarfile.TarError:
            raise serializers.ValidationError("Failed to open the chaincode tar package.")

        return value

    def validate_channel(self, value: Channel) -> Channel:
        if not value.organizations.contains(self.context["organization"]):
            raise serializers.ValidationError("You can only install chaincodes on your organization.")
        return value

    def create(self, validated_data: Dict[str, Any]) -> ChaincodeID:
        validated_data["user"] = self.context["user"]
        validated_data["organization"] = self.context["organization"]
        return ChaincodeID({"id": create_chaincode(**validated_data).id})


class ChaincodeInstallBody(ChaincodeID):
    def update(self, instance: Chaincode, validated_data: Dict[str, Any]) -> Chaincode:
        install_chaincode(
            self.context["organization"],
            instance
        )
        return instance


class ChaincodeApproveBody(ChaincodeID):
    def update(self, instance: Chaincode, validated_data: Dict[str, Any]) -> Chaincode:
        approve_chaincode(
            self.context["organization"],
            instance
        )
        return instance


class ChaincodeCommitBody(ChaincodeID):
    def update(self, instance: Chaincode, validated_data: Dict[str, Any]) -> Chaincode:
        commit_chaincode(
            self.context["organization"],
            instance
        )
        return instance


class ChaincodeRequestBody(ChaincodeID):
    action = serializers.ChoiceField(choices=[(tag.name, tag.name) for tag in ChaincodeAction])
    function = serializers.CharField()
    arguments = serializers.ListField(
        child=serializers.CharField(),
        allow_empty=True,
        required=False,
        default=[],
    )

    class Meta:
        model = Chaincode
        fields = ("id", "action", "function", "arguments")
        extra_kwargs = {
            # Temporarily make "id" writable only for validation purposes
            "id": {"read_only": False},
            "arguments": {"required": False},
        }

    def update(self, instance: Chaincode, validated_data: Dict[str, Any]) -> Chaincode:
        try:
            result = send_chaincode_request(
                self.context["organization"],
                instance,
                ChaincodeAction[validated_data["action"]],
                validated_data["function"],
                *validated_data["arguments"]
            )
            instance.result = result
            return instance
        except ChaincodeTransactionError as e:
            raise serializers.ValidationError(str(e))