<|start_filename|>cmd/kube-plex/metadata_test.go<|end_filename|>
package main

import (
	"context"
	"reflect"
	"testing"

	"github.com/go-test/deep"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes/fake"
)

func Test_pmsMetadata_FetchMetadata(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	cpuQuantity, _ := resource.ParseQuantity("1")
	validPod := corev1.Pod{
		ObjectMeta: v1.ObjectMeta{
			Namespace: "plex", Name: "pms", UID: "123",
			Annotations: map[string]string{"kube-plex/pms-addr": "service:32400", "kube-plex/mounts": "/data"},
		},
		Spec: corev1.PodSpec{
			Containers: []corev1.Container{{Name: "plex", Image: "plex:test", VolumeMounts: []corev1.VolumeMount{{Name: "data", MountPath: "/data"}}}},
			Volumes:    []corev1.Volume{{Name: "data"}},
		},
		Status: corev1.PodStatus{
			InitContainerStatuses: []corev1.ContainerStatus{{Name: "kube-plex-init", Image: "kubeplex:latest", ImageID: "kubeplex@sha256:12345"}},
			ContainerStatuses:     []corev1.ContainerStatus{{Name: "plex", Image: "pms:latest", ImageID: "pms@sha256:12345"}},
		},
	}

	tests := []struct {
		name         string
		podname      string
		podnamespace string
		pod          corev1.Pod
		wantPms      PmsMetadata
		wantErr      bool
	}{
		{"fetches info from api", "pms", "plex", validPod, PmsMetadata{
			Name: "pms", Namespace: "plex", UID: "123", PmsImage: "pms@sha256:12345", KubePlexImage: "kubeplex@sha256:12345", PmsAddr: "service:32400",
			Mounts: []string{"/data"}, VolumeMounts: validPod.Spec.Containers[0].VolumeMounts, Volumes: validPod.Spec.Volumes}, false,
		},
		{"fails on missing podname", "", "plex", validPod, PmsMetadata{}, true},
		{"fails on missing namespace", "pms", "", validPod, PmsMetadata{}, true},
		{"fails gracefully on wrong pod name", "wrong", "plex", validPod, PmsMetadata{}, true},
		{"fails gracefully on wrong namespace", "pms", "wrong", validPod, PmsMetadata{}, true},
		{"plex container missing", "pms", "plex", corev1.Pod{ObjectMeta: validPod.ObjectMeta, Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "wrong", Image: "pms:own"}}, Volumes: []corev1.Volume{{Name: "data"}, {Name: "transcode"}}}}, PmsMetadata{}, true},
		{"plex data volume missing", "pms", "plex", corev1.Pod{ObjectMeta: validPod.ObjectMeta, Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "plex", Image: "pms:own"}}, Volumes: []corev1.Volume{{Name: "transcode"}}}}, PmsMetadata{}, true},
		{"plex default volumes", "pms", "plex",
			corev1.Pod{
				ObjectMeta: v1.ObjectMeta{Namespace: "plex", Name: "pms", UID: "123", Annotations: map[string]string{"kube-plex/pms-addr": "service:32400"}},
				Spec: corev1.PodSpec{
					Containers: []corev1.Container{{Name: "plex", Image: "plex:test", VolumeMounts: []corev1.VolumeMount{{Name: "data", MountPath: "/data"}, {Name: "transcode", MountPath: "/transcode"}}}},
					Volumes:    []corev1.Volume{{Name: "transcode"}, {Name: "data"}}},
				Status: validPod.Status,
			},
			PmsMetadata{
				Name: "pms", Namespace: "plex", UID: "123", PmsImage: "pms@sha256:12345", KubePlexImage: "kubeplex@sha256:12345", PmsAddr: "service:32400",
				Mounts:       []string{"/transcode", "/data"},
				VolumeMounts: []corev1.VolumeMount{{Name: "transcode", MountPath: "/transcode"}, {Name: "data", MountPath: "/data"}},
				Volumes:      []corev1.Volume{{Name: "data"}, {Name: "transcode"}},
			},
			false},
		{"plex transcode volume missing", "pms", "plex", corev1.Pod{ObjectMeta: validPod.ObjectMeta, Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "plex", Image: "pms:own"}}, Volumes: []corev1.Volume{{Name: "data"}}}}, PmsMetadata{}, true},
		{"kube-plex debug set", "pms", "plex", corev1.Pod{
			ObjectMeta: v1.ObjectMeta{Namespace: "plex", Name: "pms", UID: "123", Annotations: map[string]string{"kube-plex/pms-addr": "a:32400", "kube-plex/loglevel": "debug", "kube-plex/mounts": ""}}, Spec: validPod.Spec, Status: validPod.Status},
			PmsMetadata{Name: "pms", Namespace: "plex", UID: "123", PmsImage: "pms@sha256:12345", KubePlexImage: "kubeplex@sha256:12345", KubePlexLevel: "debug", PmsAddr: "a:32400"},
			false,
		},
		{"renamed kube-plex container", "pms", "plex",
			corev1.Pod{ObjectMeta: v1.ObjectMeta{Namespace: "plex", Name: "pms", UID: "123", Annotations: map[string]string{"kube-plex/container-name": "kp-init", "kube-plex/pms-addr": "a:32400", "kube-plex/mounts": ""}}, Spec: validPod.Spec, Status: corev1.PodStatus{ContainerStatuses: validPod.Status.ContainerStatuses, InitContainerStatuses: []corev1.ContainerStatus{{Name: "kp-init", ImageID: "aaa@sha256:12345"}}}},
			PmsMetadata{Name: "pms", Namespace: "plex", UID: "123", PmsImage: "pms@sha256:12345", KubePlexImage: "aaa@sha256:12345", PmsAddr: "a:32400"},
			false,
		},
		{"renamed PMS container", "pms", "plex",
			corev1.Pod{ObjectMeta: v1.ObjectMeta{Namespace: "plex", Name: "pms", UID: "123", Annotations: map[string]string{"kube-plex/pms-container-name": "test", "kube-plex/pms-addr": "a:32400", "kube-plex/mounts": ""}}, Spec: validPod.Spec, Status: corev1.PodStatus{InitContainerStatuses: validPod.Status.InitContainerStatuses, ContainerStatuses: []corev1.ContainerStatus{{Name: "test", ImageID: "aaa@sha256:12345"}}}},
			PmsMetadata{Name: "pms", Namespace: "plex", UID: "123", PmsImage: "aaa@sha256:12345", KubePlexImage: "kubeplex@sha256:12345", PmsAddr: "a:32400"},
			false,
		},
		{"sets resource definitions", "pms", "plex",
			corev1.Pod{ObjectMeta: v1.ObjectMeta{Namespace: "plex", Name: "pms", UID: "123", Annotations: map[string]string{"kube-plex/pms-addr": "a:32400", "kube-plex/mounts": "", "kube-plex/resources-requests": "{\"cpu\": \"1\"}", "kube-plex/resources-limits": "{\"cpu\": \"1\"}"}}, Spec: validPod.Spec, Status: validPod.Status},
			PmsMetadata{Name: "pms", Namespace: "plex", UID: "123", PmsImage: "pms@sha256:12345", KubePlexImage: "kubeplex@sha256:12345", PmsAddr: "a:32400", ResourceRequests: corev1.ResourceList{corev1.ResourceCPU: cpuQuantity}, ResourceLimits: corev1.ResourceList{corev1.ResourceCPU: cpuQuantity}},
			false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cl := fake.NewSimpleClientset(&tt.pod)
			m, err := FetchMetadata(ctx, cl, tt.podname, tt.podnamespace)
			if (err != nil) != tt.wantErr {
				t.Errorf("pmsMetadata.FetchAPI() error = %v, wantErr %v", err, tt.wantErr)
			}
			// We don't want to check output state if error occurs
			if !tt.wantErr {
				if diff := deep.Equal(m, tt.wantPms); diff != nil {
					t.Errorf("pmsMetadata.FetchAPI() diff: %v", diff)
				}
			}
		})
	}
}

func Test_pmsMetadata_OwnerReference(t *testing.T) {
	tests := []struct {
		name    string
		obj     PmsMetadata
		want    v1.OwnerReference
		wantErr bool
	}{
		{"success", PmsMetadata{Name: "testpod", Namespace: "plex", UID: "123"}, v1.OwnerReference{APIVersion: "v1", Kind: "Pod", Name: "testpod", UID: "123"}, false},
		{"missing uuid", PmsMetadata{Name: "testpod", Namespace: "plex"}, v1.OwnerReference{}, true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			p := tt.obj
			got, err := p.OwnerReference()
			if (err != nil) != tt.wantErr {
				t.Errorf("pmsMetadata.OwnerReference() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if !reflect.DeepEqual(got, tt.want) {
				t.Errorf("pmsMetadata.OwnerReference() = %v, want %v", got, tt.want)
			}
		})
	}
}

func Test_pmsMetadata_LauncherCmd(t *testing.T) {
	tests := []struct {
		name string
		p    PmsMetadata
		args []string
		want []string
	}{
		{"generates bare cmd", PmsMetadata{PmsAddr: "a:32400"}, []string{"a"}, []string{"/shared/transcode-launcher", "--pms-addr=a:32400", "--listen=:32400", "--", "a"}},
		{"generates codec server url", PmsMetadata{PmsAddr: "a:32400", PodIP: "172.16.58.3", CodecPort: 1234}, []string{"a"}, []string{"/shared/transcode-launcher", "--pms-addr=a:32400", "--listen=:32400", "--codec-server-url=http://172.16.58.3:1234/", "--codec-dir=/shared/codecs/", "--", "a"}},
		{"generates debug flag", PmsMetadata{PmsAddr: "a:32400", KubePlexLevel: "debug"}, []string{"a"}, []string{"/shared/transcode-launcher", "--pms-addr=a:32400", "--listen=:32400", "--loglevel=debug", "--", "a"}},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			p := tt.p
			if got := p.LauncherCmd(tt.args...); !reflect.DeepEqual(got, tt.want) {
				t.Errorf("pmsMetadata.LauncherCmd() = %v, want %v", got, tt.want)
			}
		})
	}
}

func Test_getVolumesAndMounts(t *testing.T) {
	testPod := corev1.Pod{
		Spec: corev1.PodSpec{
			Containers: []corev1.Container{{
				Name: "container",
				VolumeMounts: []corev1.VolumeMount{
					{Name: "missing", MountPath: "/missing"},
					{Name: "data", MountPath: "/data"},
					{Name: "data", MountPath: "/data1", SubPath: "s1"},
					{Name: "data", MountPath: "/data2", SubPath: "s2"},
					{Name: "transcode", MountPath: "/transcode"},
					{Name: "full", MountPath: "/full", ReadOnly: true},
				},
			}},
			Volumes: []corev1.Volume{{Name: "data"}, {Name: "transcode"}, {Name: "full", VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}}},
		},
	}
	type args struct {
		dirs []string
		name string
	}
	tests := []struct {
		name       string
		args       args
		wantVolume []corev1.Volume
		wantMount  []corev1.VolumeMount
		wantErr    bool
	}{
		{"default kube-plex mounts",
			args{dirs: []string{"/data", "/transcode"}, name: "container"},
			[]corev1.Volume{{Name: "data"}, {Name: "transcode"}},
			[]corev1.VolumeMount{{Name: "data", MountPath: "/data"}, {Name: "transcode", MountPath: "/transcode"}},
			false,
		},
		{"deduplicate volumes",
			args{dirs: []string{"/data1", "/data2"}, name: "container"},
			[]corev1.Volume{{Name: "data"}},
			[]corev1.VolumeMount{{Name: "data", MountPath: "/data1", SubPath: "s1"}, {Name: "data", MountPath: "/data2", SubPath: "s2"}},
			false,
		},
		{"errors on invalid container", args{dirs: []string{"/data"}, name: "fail"}, nil, nil, true},
		{"errors on invalid path", args{dirs: []string{"/data", "/fail"}, name: "fail"}, nil, nil, true},
		// Test this even if it's an invalid case due to validation in Kubernetes
		{"errors on missing volume", args{dirs: []string{"/missing"}, name: "container"}, nil, nil, true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			volumes, mounts, err := getVolumesAndMounts(tt.args.dirs, &testPod, tt.args.name)
			if (err != nil) != tt.wantErr {
				t.Errorf("getVolumesAndMounts() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if diff := deep.Equal(volumes, tt.wantVolume); diff != nil {
				t.Errorf("getVolumesAndMounts() volumes don't match diff = %v", diff)
			}
			if diff := deep.Equal(mounts, tt.wantMount); diff != nil {
				t.Errorf("getVolumesAndMounts() mounts don't match diff = %v", diff)
			}
		})
	}
}

func Test_parseResources(t *testing.T) {
	cpuMilli, _ := resource.ParseQuantity("100m")
	qOne, _ := resource.ParseQuantity("1")
	tests := []struct {
		name    string
		t       string
		want    corev1.ResourceList
		wantErr bool
	}{
		{"empty input returns nil", "", nil, false},
		{"handles json input", "{\"cpu\": \"100m\"}", corev1.ResourceList{corev1.ResourceCPU: cpuMilli}, false},
		{"handles arbitrary fields", "{\"intel.gpu\": 1}", corev1.ResourceList{"intel.gpu": qOne}, false},
		{"returns error on broken json", "{\"", nil, true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := parseResourcesJSON(tt.t)
			if (err != nil) != tt.wantErr {
				t.Errorf("parseResources() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if !reflect.DeepEqual(got, tt.want) {
				t.Errorf("parseResources() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestPmsMetadata_ResourceRequirements(t *testing.T) {
	cpuMilli, _ := resource.ParseQuantity("100m")
	type fields struct {
		ResourceRequests corev1.ResourceList
		ResourceLimits   corev1.ResourceList
	}
	tests := []struct {
		name   string
		fields fields
		want   corev1.ResourceRequirements
	}{
		{"empty object when no requests exist", fields{}, corev1.ResourceRequirements{}},
		{"limit only", fields{ResourceRequests: corev1.ResourceList{corev1.ResourceCPU: cpuMilli}}, corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: cpuMilli}}},
		{"request only", fields{ResourceLimits: corev1.ResourceList{corev1.ResourceCPU: cpuMilli}}, corev1.ResourceRequirements{Limits: corev1.ResourceList{corev1.ResourceCPU: cpuMilli}}},
		{"both requests and limits",
			fields{ResourceLimits: corev1.ResourceList{corev1.ResourceCPU: cpuMilli}, ResourceRequests: corev1.ResourceList{corev1.ResourceCPU: cpuMilli}},
			corev1.ResourceRequirements{Limits: corev1.ResourceList{corev1.ResourceCPU: cpuMilli}, Requests: corev1.ResourceList{corev1.ResourceCPU: cpuMilli}},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			m := PmsMetadata{
				ResourceRequests: tt.fields.ResourceRequests,
				ResourceLimits:   tt.fields.ResourceLimits,
			}
			if got := m.ResourceRequirements(); !reflect.DeepEqual(got, tt.want) {
				t.Errorf("PmsMetadata.ResourceRequirements() = %v, want %v", got, tt.want)
			}
		})
	}
}

func Test_getContainerImage(t *testing.T) {
	type args struct {
		annotation string
		defname    string
		pod        *corev1.Pod
		status     []corev1.ContainerStatus
	}
	tests := []struct {
		name      string
		args      args
		wantImage string
		wantName  string
		wantErr   bool
	}{
		{"docker pullable", args{defname: "kube-plex", pod: &corev1.Pod{}, status: []corev1.ContainerStatus{corev1.ContainerStatus{Name: "kube-plex", ImageID: "docker-pullable://a/b@sha256:abc"}}}, "a/b@sha256:abc", "kube-plex", false},
		{"containerd image", args{defname: "kube-plex", pod: &corev1.Pod{}, status: []corev1.ContainerStatus{corev1.ContainerStatus{Name: "kube-plex", ImageID: "a/b@sha256:abc"}}}, "a/b@sha256:abc", "kube-plex", false},
		{"image in annotation", args{annotation: "a", defname: "none", pod: &corev1.Pod{ObjectMeta: v1.ObjectMeta{Annotations: map[string]string{"a": "kubeplex"}}}, status: []corev1.ContainerStatus{corev1.ContainerStatus{Name: "kubeplex", ImageID: "a/b@sha256:abc"}}}, "a/b@sha256:abc", "kubeplex", false},
		{"name mismatch", args{defname: "kube-plex", pod: &corev1.Pod{}, status: []corev1.ContainerStatus{corev1.ContainerStatus{Name: "kubeplex", ImageID: "a/b@sha256:abc"}}}, "", "", true},
		// TODO: Add test cases.
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			gotImage, gotName, err := getContainerImage(tt.args.annotation, tt.args.defname, tt.args.pod, tt.args.status)
			if (err != nil) != tt.wantErr {
				t.Errorf("getContainerImage() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if gotImage != tt.wantImage {
				t.Errorf("getContainerImage() got image = %v, want %v", gotImage, tt.wantImage)
			}
			if gotName != tt.wantName {
				t.Errorf("getContainerImage() got name = %v, want %v", gotName, tt.wantName)
			}
		})
	}
}


<|start_filename|>cmd/kube-plex/main_test.go<|end_filename|>
package main

import (
	"os"
	"testing"

	"github.com/go-logr/logr"
	"k8s.io/klog/v2"
)

// EmptyLogger implements logr.Logging
type EmptyLogger struct{}

// Enabled always returns false
func (e *EmptyLogger) Enabled() bool {
	return false
}

// Info does nothing
func (e *EmptyLogger) Info(msg string, keysAndValues ...interface{}) {}

// Error does nothing
func (e *EmptyLogger) Error(err error, msg string, keysAndValues ...interface{}) {}

// V returns itself
func (e *EmptyLogger) V(level int) logr.Logger {
	return e
}

// WithValues returns itself
func (e *EmptyLogger) WithValues(keysAndValues ...interface{}) logr.Logger {
	return e
}

// WithName returns itself
func (e *EmptyLogger) WithName(name string) logr.Logger {
	return e
}

// disable logging in all tests
func TestMain(m *testing.M) {
	klog.SetLogger(&EmptyLogger{})
	os.Exit(m.Run())
}

func Test_needBypass(t *testing.T) {
	tests := []struct {
		name string
		args []string
		want bool
	}{
		{"bypass aec3_eae", []string{"...", "-codec:1", "eac3_eae", "-eaeprefix:1", "..."}, true},
		{"bypass ac3_eae", []string{"...", "-codec:1", "ac3_eae", "-eaeprefix:1", "..."}, true},
		{"bypass truehd_eae", []string{"...", "-codec:1", "truehd_eae", "-eaeprefix:1", "..."}, true},
		{"bypass mlp_eae", []string{"...", "-codec:1", "mlp_eae", "-eaeprefix:1", "..."}, true},
		{"don't bypass with ac3", []string{"...", "-codec:1", "ac3", "-prefix:1", "..."}, false},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := needBypass(tt.args); got != tt.want {
				t.Errorf("needBypass() = %v, want %v", got, tt.want)
			}
		})
	}
}


<|start_filename|>cmd/transcode-launcher/codecs.go<|end_filename|>
package main

import (
	"archive/tar"
	"fmt"
	"io"
	"net/http"
	"os"
	"path/filepath"

	"k8s.io/klog/v2"
)

func downloadCodecs(path, url string) error {
	err := os.MkdirAll(path, 0777)
	if err != nil {
		return fmt.Errorf("failed to create codec directory: %v", err)
	}

	res, err := http.Get(url)
	if err != nil {
		return fmt.Errorf("error when fetching codec package: %v", err)
	}

	// ensure that res.Body is closed
	defer res.Body.Close()
	return unpackCodecs(path, res.Body)
}

func unpackCodecs(dest string, r io.Reader) error {
	klog.Infof("Unpacking codecs to: %s", dest)
	tr := tar.NewReader(r)
	for {
		hdr, err := tr.Next()
		if err == io.EOF {
			break
		}
		if err != nil {
			return fmt.Errorf("invalid tar header: %v", err)
		}

		finfo := hdr.FileInfo()
		name := hdr.Name
		absName := filepath.Join(dest, name)
		if finfo.IsDir() {
			if err := os.MkdirAll(absName, finfo.Mode()); err != nil {
				return fmt.Errorf("failed to create directory %s: %v", name, err)
			}
			klog.V(1).Infof("Mkdir(%s)", name)
			continue
		}

		file, err := os.OpenFile(
			absName,
			os.O_RDWR|os.O_CREATE|os.O_TRUNC,
			finfo.Mode().Perm(),
		)
		if err != nil {
			return fmt.Errorf("error creating file %s: %v", name, err)
		}

		n, cErr := io.Copy(file, tr)
		if err := file.Close(); err != nil {
			return fmt.Errorf("error while closing file %s: %v", name, err)
		}
		if cErr != nil {
			return fmt.Errorf("error while writing to file %s: %v", name, err)
		}
		if n != finfo.Size() {
			return fmt.Errorf("wrote %d, size %d", n, finfo.Size())
		}
		klog.V(1).Infof("Write(%s)", name)
	}
	return nil
}


<|start_filename|>cmd/transcode-launcher/copystream.go<|end_filename|>
package main

import (
	"context"
	"fmt"
	"io"
	"net"

	"k8s.io/klog/v2"
)

// Convenience wrapper for listening on a given port and launcing dialAndCopy() for every
// incoming connection
func copyListener(ctx context.Context, listenAddr, serverAddr string) error {
	l, err := net.Listen("tcp", listenAddr)
	if err != nil {
		return fmt.Errorf("failed to listen on %s: %v", listenAddr, err)
	}
	defer l.Close()

	for {
		cConn, err := l.Accept()
		if err != nil {
			return fmt.Errorf("Accept() returned an error: %v", err)
		}
		go dialAndCopy(ctx, cConn, serverAddr)
	}
}

// dialAndCopy is a naive tunnel between 2 connections. It copies input and output between
// The client and server. Any errors will close the connection
func dialAndCopy(ctx context.Context, cConn net.Conn, addr string) {
	// Close client connection once we are done
	defer cConn.Close()

	var d net.Dialer
	sConn, err := d.DialContext(ctx, "tcp", addr)
	if err != nil {
		klog.Exitf("Dial() failed: %v", err)
	}
	defer sConn.Close()
	inErr := make(chan error, 1)
	outErr := make(chan error, 1)

	go func() {
		_, err := io.Copy(sConn, cConn)
		inErr <- err
	}()

	go func() {
		_, err := io.Copy(cConn, sConn)
		outErr <- err
	}()

	select {
	case err = <-inErr:
		if err != nil {
			klog.Infof("error while reading from client: %v", err)
		}
	case err = <-outErr:
		if err != nil {
			klog.Infof("error while reading from server: %v", err)
		}
	case <-ctx.Done():
		klog.Infof("context done")
	}
}


<|start_filename|>Dockerfile<|end_filename|>
FROM golang:1.16 AS builder

WORKDIR /go/src/app
COPY . .

RUN CGO_ENABLED=0 go build -o ./ ./cmd/...

FROM alpine:3.13

COPY --from=builder /go/src/app/kube-plex /go/src/app/transcode-launcher /

<|start_filename|>cmd/kube-plex/metadata.go<|end_filename|>
package main

import (
	"context"
	"fmt"
	"sort"
	"strings"

	corev1 "k8s.io/api/core/v1"
	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
)

const (
	pmsURL                = "kube-plex/pms-addr"
	pmsContainer          = "kube-plex/pms-container-name"
	pmsMounts             = "kube-plex/mounts"
	kubePlexLevel         = "kube-plex/loglevel"
	kubePlexContainer     = "kube-plex/container-name"
	kubePlexResourceReq   = "kube-plex/resources-requests"
	kubePlexResourceLimit = "kube-plex/resources-limits"
)

// PmsMetadata describes a Plex Media Server instance running in kubernetes.
type PmsMetadata struct {
	Name             string               // Pod Name
	Namespace        string               // Pod Namespace
	UID              types.UID            // Pod UID
	PodIP            string               // Pod IP address
	Mounts           []string             // List of mounts (paths) to copy to transcoder
	VolumeMounts     []corev1.VolumeMount // kube-plex volume mounts
	Volumes          []corev1.Volume      // kube-plex needed volumes
	ResourceRequests corev1.ResourceList  // Resource requests definition for kube-plex
	ResourceLimits   corev1.ResourceList  // Resource limits definition for kube-plex
	KubePlexImage    string               // container image for kube-plex
	KubePlexLevel    string               // loglevel of kubeplex processes
	CodecPort        int                  // port on which the codec service runs
	PmsImage         string               // container image used by Plex Media Server
	PmsAddr          string               // URL for Plex Media Server
}

// FetchMetadata fetches and populates a metadata object based on the current environment
func FetchMetadata(ctx context.Context, cl kubernetes.Interface, name, namespace string) (PmsMetadata, error) {
	if name == "" {
		return PmsMetadata{}, fmt.Errorf("pod name is empty")
	}

	if namespace == "" {
		return PmsMetadata{}, fmt.Errorf("namespace is empty")
	}

	pod, err := cl.CoreV1().Pods(namespace).Get(ctx, name, v1.GetOptions{})
	if err != nil {
		return PmsMetadata{}, fmt.Errorf("unable to fetch Pod info: %v", err)
	}

	m := PmsMetadata{
		Name:      pod.GetName(),
		Namespace: pod.GetNamespace(),
		UID:       pod.GetUID(),
		PodIP:     pod.Status.PodIP,
	}

	// Get PMS URL
	a := pod.GetAnnotations()
	u, ok := a[pmsURL]
	if !ok {
		return PmsMetadata{}, fmt.Errorf("unable to determine plex service URL")
	}
	m.PmsAddr = u

	// Get debugging status
	d := a[kubePlexLevel]
	// TODO: It would be nice to enforce all valid options here
	m.KubePlexLevel = d

	// Plex media server container image
	pmsimage, pmsname, err := getContainerImage(pmsContainer, "plex", pod, pod.Status.ContainerStatuses)
	if err != nil {
		return PmsMetadata{}, fmt.Errorf("unable to determine Plex Media server image (set container name with '%s' annotation): %v", pmsContainer, err)
	}
	m.PmsImage = pmsimage

	// Kube-Plex container image
	kpimage, _, err := getContainerImage(kubePlexContainer, "kube-plex-init", pod, pod.Status.InitContainerStatuses)
	if err != nil {
		return PmsMetadata{}, fmt.Errorf("unable to determine kube-plex image (set init-container name with '%s' annotation): %v", kubePlexContainer, err)
	}
	m.KubePlexImage = kpimage

	// mounts to copy over
	mlist, ok := a[pmsMounts]
	if !ok {
		// default value, matches the old behaviour
		mlist = "/transcode,/data"
	}
	if mlist != "" {
		m.Mounts = strings.Split(mlist, ",")
	}

	v, vm, err := getVolumesAndMounts(m.Mounts, pod, pmsname)
	if err != nil {
		return PmsMetadata{}, fmt.Errorf("failed to get volumes and mounts: %v", err)
	}
	m.VolumeMounts = vm
	m.Volumes = v

	// resource requests and limits
	r := a[kubePlexResourceReq]
	rl, err := parseResourcesJSON(r)
	if err != nil {
		return PmsMetadata{}, fmt.Errorf("failed to parse resource requests `%s`: %v", r, err)
	}
	m.ResourceRequests = rl

	l := a[kubePlexResourceLimit]
	ll, err := parseResourcesJSON(l)
	if err != nil {
		return PmsMetadata{}, fmt.Errorf("failed to parse resource limits `%s`: %v", l, err)
	}
	m.ResourceLimits = ll

	return m, nil
}

// ResourceRequirements creates a container resource requirements object by combining limits and requests
func (p PmsMetadata) ResourceRequirements() corev1.ResourceRequirements {
	return corev1.ResourceRequirements{
		Limits:   p.ResourceLimits,
		Requests: p.ResourceRequests,
	}
}

// getContainerImage from pod status based on the annotation given
func getContainerImage(annotation, defname string, pod *corev1.Pod, status []corev1.ContainerStatus) (string, string, error) {
	a := pod.GetAnnotations()
	name, ok := a[annotation]
	if !ok {
		name = defname
	}
	for _, c := range status {
		if c.Name == name {
			imageID := c.ImageID
			if strings.HasPrefix(imageID, "docker-pullable://") {
				imageID = imageID[18:]
			}
			return imageID, name, nil
		}
	}
	return "", "", fmt.Errorf("no containers found by name %s", name)
}

// getVolumesAndMounts for given directories in the pod
func getVolumesAndMounts(dirs []string, pod *corev1.Pod, name string) ([]corev1.Volume, []corev1.VolumeMount, error) {
	if len(dirs) == 0 {
		return nil, nil, nil
	}

	var c *corev1.Container
	for _, d := range pod.Spec.Containers {
		if d.Name == name {
			c = &d
		}
	}
	if c == nil {
		return nil, nil, fmt.Errorf("container %s not found in pod", name)
	}

	mounts := map[string]corev1.VolumeMount{}
	for _, m := range c.VolumeMounts {
		mounts[m.MountPath] = m
	}

	volumes := map[string]corev1.Volume{}
	for _, v := range pod.Spec.Volumes {
		volumes[v.Name] = v
	}

	// since a single volume can be mounted multiple times, deduplicate volumes
	var vm []corev1.VolumeMount
	vtmp := map[string]corev1.Volume{}
	for _, d := range dirs {
		m, ok := mounts[d]
		if !ok {
			return nil, nil, fmt.Errorf("no volume mount defined for '%s'", d)
		}
		vm = append(vm, m)

		v, ok := volumes[m.Name]
		if !ok {
			return nil, nil, fmt.Errorf("no volume definition found for volume '%s'", m.Name)
		}
		vtmp[m.Name] = v
	}

	// Finally collect the volumes in a slice for returning (in a predictable order for testing)
	vkeys := make([]string, 0, len(vtmp))
	for k := range vtmp {
		vkeys = append(vkeys, k)
	}
	sort.Strings(vkeys)

	var v []corev1.Volume
	for _, k := range vkeys {
		v = append(v, vtmp[k])
	}

	return v, vm, nil
}

// OwnerReference creates an owner reference that can be used to trigger cleanup
// when this PMS instance is deleted
func (p PmsMetadata) OwnerReference() (v1.OwnerReference, error) {
	if p.UID == "" {
		return v1.OwnerReference{}, fmt.Errorf("UUID is empty, has Fetch() been run?")
	}

	return v1.OwnerReference{
		APIVersion: "v1",
		Kind:       "Pod",
		Name:       p.Name,
		UID:        p.UID,
	}, nil
}

// LauncherCmd returns a valid launcher command for this transcode operation
func (p PmsMetadata) LauncherCmd(args ...string) []string {
	a := []string{
		"/shared/transcode-launcher",
		fmt.Sprintf("--pms-addr=%s", p.PmsAddr),
		"--listen=:32400",
	}
	if p.CodecPort != 0 {
		a = append(a,
			fmt.Sprintf("--codec-server-url=http://%s:%d/", p.PodIP, p.CodecPort),
			"--codec-dir=/shared/codecs/",
		)
	}
	if p.KubePlexLevel != "" {
		a = append(a, fmt.Sprintf("--loglevel=%s", p.KubePlexLevel))
	}
	a = append(a, "--")
	return append(a, args...)
}

// podTemplate is a dummy pod definition that is used to construct a parseable
// resource for Kubernetes client parser.
const specTemplate = `
apiVersion: v1
kind: Pod
spec:
  containers:
  - resources:
      requests:
        `

// parseResourceJSON uses Kubernetes parser to parse a resource limits or requests definition to a ResourceList
func parseResourcesJSON(t string) (corev1.ResourceList, error) {
	if t == "" {
		return nil, nil
	}

	o, _, err := scheme.Codecs.UniversalDeserializer().Decode([]byte(specTemplate+t), nil, nil)
	if err != nil {
		return nil, fmt.Errorf("unable to read resource yaml: %v", err)
	}

	p := o.(*corev1.Pod)
	c := p.Spec.Containers
	if len(c) != 1 {
		return nil, fmt.Errorf("invalid object returned by decode")
	}
	return c[0].Resources.Requests, nil
}


<|start_filename|>cmd/kube-plex/main_windows.go<|end_filename|>
package main

import "os"

// Only signal to catch in windows is os.Interrupt
var shutdownSignals = []os.Signal{os.Interrupt}

// protectSigKill is a no-op on Windows
func protectSigKill(ctx context.Context) context.Context { return ctx }


<|start_filename|>cmd/kube-plex/codecserve_test.go<|end_filename|>
package main

import (
	"archive/tar"
	"io"
	"net/http"
	"net/http/httptest"
	"reflect"
	"testing"
	"testing/fstest"
)

func Test_codecPackage(t *testing.T) {
	tests := []struct {
		name     string
		fs       fstest.MapFS
		contents []string
	}{
		{"packages a directory", fstest.MapFS{"foo/bar": &fstest.MapFile{Data: []byte("bar")}}, []string{"foo", "foo/bar"}},
		{"handles empty diretory", fstest.MapFS{}, []string{}},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			c := codecServe{fs: tt.fs}
			ts := httptest.NewServer(http.HandlerFunc(c.codecPackage))
			defer ts.Close()

			res, err := http.Get(ts.URL)
			if err != nil {
				t.Fatalf("codecPackage() HTTP GET err = %v", err)
			}

			if res.StatusCode != http.StatusOK {
				t.Errorf("codecPackage() returned HTTP error: %v", res.Status)
			}

			if res.Header.Get("Content-Type") != "application/x-tar" {
				t.Errorf("codecPackage() returned incorrect content type: %v", res.Header.Get("Content-Type"))
			}

			tr := tar.NewReader(res.Body)
			defer res.Body.Close()

			files := []string{}
			for {
				hdr, err := tr.Next()
				if err == io.EOF {
					break
				}
				if err != nil {
					t.Fatalf("error reading tar stream: %v", err)
				}
				files = append(files, hdr.Name)
			}

			if !reflect.DeepEqual(tt.contents, files) {
				t.Errorf("codecPackage() contents differ, want = %v, got = %v", tt.contents, files)
			}

		})
	}
}


<|start_filename|>internal/ffmpeg/ffmpeg_test.go<|end_filename|>
// ffmpeg package contains helpers to unescape and escape FFmpeg strings
package ffmpeg

import (
	"testing"
)

func TestUnescape(t *testing.T) {
	tests := []struct {
		name string
		in   string
		want string
	}{
		{"string that needs no escaping", "lorem ipsum", "lorem ipsum"},
		{"escaped with backslash", "lorem\\ ipsum", "lorem ipsum"},
		{"escaped with quotes", "'lorem ipsum'", "lorem ipsum"},
		{"backslash within quotes", "'lorem\\ ipsum'", "lorem\\ ipsum"},
		{"different escapes in same", "lorem\\ ipsum 'lorem ipsum'", "lorem ipsum lorem ipsum"},
		{"escaped quotes", "lorem\\' ipsum\\'", "lorem' ipsum'"},
		{"mixed escapes", "lorem\\' ip'su\\'m", "lorem' ipsu\\m"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := Unescape(tt.in); got != tt.want {
				t.Errorf("Unescape() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestEscape(t *testing.T) {
	tests := []struct {
		name string
		in   string
		want string
	}{
		{"escapes backslash", "lorem\\ipsum", "lorem\\\\ipsum"},
		{"escapes quotation marks", "lorem'ipsum", "lorem\\'ipsum"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := Escape(tt.in); got != tt.want {
				t.Errorf("Escape() = %v, want %v", got, tt.want)
			}
		})
	}
}


<|start_filename|>internal/logger/plex.go<|end_filename|>
// Package logger provides a logger which will write log entries to Plex Media Server
package logger

import (
	"fmt"
	"net/http"
	"net/url"
	"os"
	"strings"

	"github.com/go-logr/logr"
)

// Log levels known by Plex
const (
	PlexLogError = iota
	PlexLogWarning
	PlexLogInfo
	PlexLogDebug
	PlexLogVerbose
)

// NewPlexLogger returns a PlexLogger instance that has URL preset
//
// URL should be the base url for Plex Media Server, `/log` path will be added
func NewPlexLogger(name, token, plexurl string) (*PlexLogger, error) {
	u, err := url.Parse(plexurl)
	if err != nil {
		return nil, fmt.Errorf("unable to parse url: %v", err)
	}

	u.Path = strings.TrimSuffix(u.Path, "/") + "/log"

	return &PlexLogger{
		plexURL:   u,
		plexToken: token,
		name:      name,
	}, nil
}

// PlexLogger is a single instance of Plex which is used for logging
type PlexLogger struct {
	plexURL   *url.URL // Plex url, includes plex source. (http://127.0.0.1:32400/?source=Transcoder)
	plexToken string   // Plex token for authentication

	name      string
	keyValues map[string]interface{}
}

// Enabled tests whether this Logger is enabled.
// For now we assume that the logger is always enabled
func (*PlexLogger) Enabled() bool {
	return true
}

// Info level logs are written directly to Plex
func (l *PlexLogger) Info(msg string, kvs ...interface{}) {
	l.send(PlexLogInfo, msg, kvs...)
}

// Error logs will have the error message passed as error key
func (l *PlexLogger) Error(err error, msg string, kvs ...interface{}) {
	if err != nil {
		kvs = append(kvs, "error", err)
	}
	l.send(PlexLogError, msg, kvs...)
}

// V returns a logger with the given verbosity
//
// In practice, this only adds a verbosity label to the log entry
func (l *PlexLogger) V(v int) logr.Logger {
	return l.WithValues("verbosity", v)
}

// WithName adds an element to the logger name
func (l *PlexLogger) WithName(name string) logr.Logger {
	return &PlexLogger{
		plexURL:   l.plexURL,
		plexToken: l.plexToken,
		name:      l.name + "." + name,
		keyValues: l.keyValues,
	}
}

// WithValues adds key value pairs to the logger
func (l *PlexLogger) WithValues(kvs ...interface{}) logr.Logger {
	newMap := make(map[string]interface{}, len(l.keyValues)+len(kvs)/2)
	for k, v := range l.keyValues {
		newMap[k] = v
	}
	for i := 0; i < len(kvs); i += 2 {
		newMap[kvs[i].(string)] = kvs[i+1]
	}

	return &PlexLogger{
		plexURL:   l.plexURL,
		plexToken: l.getPlexToken(),
		name:      l.name,
		keyValues: newMap,
	}
}

// send message to PMS. Wrap all key value pairs to a text string since Plex has
// no concept of metadata other than message level.
//
// The request includes Plex token if it's available through the environment
func (l *PlexLogger) send(level int, msg string, kvs ...interface{}) {
	kvmsg := []string{}
	for k, v := range l.keyValues {
		kvmsg = append(kvmsg, fmt.Sprintf("%s:%+v", k, v))
	}
	for i := 0; i < len(kvs); i += 2 {
		kvmsg = append(kvmsg, fmt.Sprintf("%s:%+v", kvs[i], kvs[i+1]))
	}

	if len(kvmsg) > 0 {
		msg = fmt.Sprintf("%s %+v", msg, kvmsg)
	}

	u := l.getURL()
	q := u.Query()
	q.Set("level", fmt.Sprintf("%d", level))
	q.Set("message", msg)
	q.Set("source", l.name)
	u.RawQuery = q.Encode()

	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
	if err != nil {
		// We have an error, but no place to report it. Bail out!
		return
	}

	plexToken := l.getPlexToken()
	if plexToken != "" {
		req.Header.Add("X-Plex-Token", plexToken)
	}
	req.Header.Add("User-Agent", "PlexLogger")

	// Ignore results
	_, err = http.DefaultClient.Do(req)
	if err != nil {
		fmt.Printf("ERROR! %v", err)
	}
}

// getURL returns either the set URL or the default URL if unset
func (l *PlexLogger) getURL() url.URL {
	if l.plexURL == nil {
		u, _ := url.Parse("http://1172.16.17.32:32400/log")
		return *u
	}
	return *l.plexURL
}

// getPlexToken returns the plex token from struct if it exists or falls back to
// X_PLEX_TOKEN environment variable
func (l *PlexLogger) getPlexToken() string {
	if l.plexToken != "" {
		return l.plexToken
	}
	return os.Getenv("X_PLEX_TOKEN")
}


<|start_filename|>internal/ffmpeg/ffmpeg.go<|end_filename|>
// ffmpeg package contains helpers to unescape and escape FFmpeg strings
package ffmpeg

import "regexp"

// Unescape takes a string that has FFmpeg compatible escape characters and returns the raw string
func Unescape(s string) string {
	re := regexp.MustCompile(`('(?P<s>[^']*)'|\\(?P<c>.))`)
	return string(re.ReplaceAll([]byte(s), []byte("${s}${c}")))
}

// Escape adds FFmpeg compatible escaping to a string
func Escape(s string) string {
	re := regexp.MustCompile(`(\\|')`)
	return string(re.ReplaceAll([]byte(s), []byte("\\$1")))
}


<|start_filename|>cmd/kube-plex/main_posix.go<|end_filename|>
package main

import (
	"context"
	"fmt"
	"os"
	"os/exec"
	"strconv"
	"syscall"
	"time"
)

// Catch interrupt and SIGTERM and terminate gracefully
var shutdownSignals = []os.Signal{os.Interrupt, syscall.SIGTERM}

// ProtectSigKill takes measures to shield transcoding process from SIGKILL
func protectSigKill(ctx context.Context) context.Context {
	// Verify that we aren't already running with protections
	protected, _ := strconv.ParseBool(os.Getenv("KUBEPLEX_SIGKILL_PROTECTION"))
	if !protected {
		// Prepare for self exec
		args := os.Args
		cmd := exec.Command(args[0], args[1:]...)
		cmd.Stdout = os.Stdout
		cmd.Stderr = os.Stderr

		// Set environment variables for signalling
		cmd.Env = append(os.Environ(),
			"KUBEPLEX_SIGKILL_PROTECTION=true",
			fmt.Sprintf("KUBEPLEX_SIGKILL_PARENT_PID=%d", os.Getpid()),
		)

		// Run command
		err := cmd.Run()

		// If child process returned an error, set error code accordingly
		ecode := 0
		if err != nil {
			if cerr, ok := err.(*exec.ExitError); ok {
				ecode = cerr.ExitCode()
			}
			fmt.Printf("Protected process returned an error: %v\n", err)
		}

		// Quit once subprocess is done
		os.Exit(ecode)
	}

	ppvar := os.Getenv("KUBEPLEX_SIGKILL_PARENT_PID")
	if ppvar == "" {
		fmt.Printf("Parent PID not provided in KUBEPLEX_SIGKILL_PARENT_PID!")
		os.Exit(1)
	}

	ppid, err := strconv.Atoi(ppvar)
	if err != nil {
		fmt.Printf("Invalid parent pid %v: %v", ppvar, err)
		os.Exit(1)
	}

	// FindProcess always returns an os.Process{}
	parent, _ := os.FindProcess(ppid)

	// Set up monitoring for parent process. This is necessary since the parent
	// is killed with SIGKILL and we don't get any signals for this.
	ctx, cancel := context.WithCancel(ctx)
	go func() {
		ticker := time.NewTicker(1 * time.Second)
		defer ticker.Stop()
		defer cancel()

		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				err := parent.Signal(syscall.Signal(0))
				if err == os.ErrProcessDone {
					return
				}
			}
		}
	}()

	// Cleanup environment set specifically for the protections
	os.Unsetenv("KUBEPLEX_SIGKILL_PROTECTION")
	os.Unsetenv("KUBEPLEX_SIGKILL_PARENT_PID")

	return ctx
}


<|start_filename|>cmd/kube-plex/codecserve.go<|end_filename|>
package main

import (
	"archive/tar"
	"fmt"
	"io"
	"io/fs"
	"net"
	"net/http"
	"os"

	"k8s.io/klog/v2"
)

type codecServe struct {
	fs fs.FS
}

func (c codecServe) codecPackage(w http.ResponseWriter, r *http.Request) {
	w.Header().Add("Content-type", "application/x-tar")
	tw := tar.NewWriter(w)
	err := fs.WalkDir(c.fs, ".", func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			return fmt.Errorf("received error from WalkDir: %v", err)
		}

		// Don't add the parent directory to tar
		if path == "." {
			return nil
		}

		f, err := d.Info()
		if err != nil {
			return fmt.Errorf("file info read failed: %v", err)
		}

		hdr, err := tar.FileInfoHeader(f, f.Name())
		if err != nil {
			return fmt.Errorf("unable to create tar header for file %s: %v", f.Name(), err)
		}

		// Use relative path as the filename, since tar wants a parent directories for the entry
		hdr.Name = path

		if err := tw.WriteHeader(hdr); err != nil {
			return fmt.Errorf("tar header write failed: %v", err)
		}

		// for directories, we only need the header
		if d.IsDir() {
			return nil
		}

		fr, err := c.fs.Open(path)
		if err != nil {
			return fmt.Errorf("could not open file %s: %v", d.Name(), err)
		}
		defer fr.Close()

		if _, err := io.Copy(tw, fr); err != nil {
			return fmt.Errorf("writing to tar failed: %v", err)
		}
		return nil
	})
	if err != nil {
		klog.Errorf("failed to create tar package: %v", err)
	}
}

func startCodecServe(path string, l net.Listener) error {
	fp := os.DirFS(path)
	f := codecServe{fs: fp}
	http.HandleFunc("/", f.codecPackage)
	return http.Serve(l, nil)
}


<|start_filename|>internal/logger/plex_test.go<|end_filename|>
// Package logger provides a logger which will write log entries to Plex Media Server
package logger

import (
	"net/http"
	"net/http/httptest"
	"net/url"
	"reflect"
	"strconv"
	"testing"
)

func TestPlexLogger_getURL(t *testing.T) {
	l := &PlexLogger{}
	want, _ := url.Parse("http://127.0.0.1:32400/log")

	if got := l.getURL(); !reflect.DeepEqual(got, *want) {
		t.Errorf("PlexLogger.getURL() when empty = %v, want %v", got, *want)
	}

	want, _ = url.Parse("http://test:1234/log")
	l = &PlexLogger{plexURL: want}

	got := l.getURL()
	if !reflect.DeepEqual(got, *want) {
		t.Errorf("PlexLogger.getURL() when plexURL set = %v, want %v", got, *want)
	}
}

func TestPlexLogger_send(t *testing.T) {
	type args struct {
		level int
		msg   string
		kvs   []interface{}
	}
	tests := []struct {
		name      string
		keyValues map[string]interface{}
		token     string
		args      args
		level     int
		message   string
	}{
		{"plain message is sent", map[string]interface{}{}, "PTOKEN", args{level: PlexLogInfo, msg: "test"}, PlexLogInfo, "test"},
		{"level is respected", map[string]interface{}{}, "PTOKEN", args{level: PlexLogError, msg: "test"}, PlexLogError, "test"},
		{"message is sent when there is no token", map[string]interface{}{}, "", args{level: PlexLogInfo, msg: "test"}, PlexLogInfo, "test"},
		{"struct kv is sent", map[string]interface{}{"key": "value", "key2": "value2"}, "PTOKEN", args{level: PlexLogInfo, msg: "test"}, PlexLogInfo, "test [key:value key2:value2]"},
		{"message args are sent", map[string]interface{}{}, "PTOKEN", args{level: PlexLogInfo, msg: "test", kvs: []interface{}{"key", 1, "key2", "value2"}}, PlexLogInfo, "test [key:1 key2:value2]"},
		{"message args and struct kv are sent", map[string]interface{}{"skey": 2, "skey2": "svalue2"}, "PTOKEN", args{level: PlexLogInfo, msg: "test", kvs: []interface{}{"key", 1, "key2", "value2"}}, PlexLogInfo, "test [skey:2 skey2:svalue2 key:1 key2:value2]"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				if r.Method != http.MethodGet {
					t.Errorf("PlexLogger.send() sent %v want %v", r.Method, http.MethodGet)
				}

				msg := r.URL.Query().Get("message")
				if msg != tt.message {
					t.Errorf("PlexLogger.send() received message = %v want = %v", msg, tt.message)
				}

				token := r.Header.Get("X-Plex-Token")
				if token != tt.token {
					t.Errorf("PlexLogger.send() token header mismatch, got = %v want = %v", token, tt.token)
				}

				level := r.URL.Query().Get("level")
				l, err := strconv.Atoi(level)
				if err != nil {
					t.Errorf("PlexLogger.send() sent invalid level = %v: %v", level, err)
					return
				}

				if l != tt.level {
					t.Errorf("PlexLogger.send() received with level = %v want = %v", level, tt.level)
				}
			}))
			defer ts.Close()

			u, _ := url.Parse(ts.URL)

			l := &PlexLogger{
				plexURL:   u,
				plexToken: tt.token,
				keyValues: tt.keyValues,
				name:      "test",
			}
			l.send(tt.args.level, tt.args.msg, tt.args.kvs...)
		})
	}
}
